summaryrefslogtreecommitdiff
path: root/plugins
diff options
context:
space:
mode:
Diffstat (limited to 'plugins')
-rw-r--r--plugins/http/plugin.go27
-rw-r--r--plugins/informer/interface.go6
-rw-r--r--plugins/informer/plugin.go6
-rw-r--r--plugins/informer/rpc.go15
-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.go106
-rw-r--r--plugins/service/process.go153
-rw-r--r--plugins/static/plugin.go2
10 files changed, 330 insertions, 24 deletions
diff --git a/plugins/http/plugin.go b/plugins/http/plugin.go
index 86fcb329..8c8a86b4 100644
--- a/plugins/http/plugin.go
+++ b/plugins/http/plugin.go
@@ -17,6 +17,7 @@ import (
endure "github.com/spiral/endure/pkg/container"
"github.com/spiral/errors"
"github.com/spiral/roadrunner/v2/pkg/pool"
+ "github.com/spiral/roadrunner/v2/pkg/process"
"github.com/spiral/roadrunner/v2/pkg/worker"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/plugins/http/attributes"
@@ -332,8 +333,24 @@ func (s *Plugin) ServeHTTP(w http.ResponseWriter, r *http.Request) {
s.RUnlock()
}
-// Workers returns associated pool workers
-func (s *Plugin) Workers() []worker.BaseProcess {
+// Workers returns slice with the process states for the workers
+func (s *Plugin) Workers() []process.State {
+ workers := s.pool.Workers()
+
+ ps := make([]process.State, 0, len(workers))
+ for i := 0; i < len(workers); i++ {
+ state, err := process.WorkerProcessState(workers[i])
+ if err != nil {
+ return nil
+ }
+ ps = append(ps, state)
+ }
+
+ return ps
+}
+
+// internal
+func (s *Plugin) workers() []worker.BaseProcess {
return s.pool.Workers()
}
@@ -395,7 +412,7 @@ func (s *Plugin) AddMiddleware(name endure.Named, m Middleware) {
// Status return status of the particular plugin
func (s *Plugin) Status() status.Status {
- workers := s.Workers()
+ workers := s.workers()
for i := 0; i < len(workers); i++ {
if workers[i].State().IsActive() {
return status.Status{
@@ -409,9 +426,9 @@ func (s *Plugin) Status() status.Status {
}
}
-// Status return status of the particular plugin
+// Ready return readiness status of the particular plugin
func (s *Plugin) Ready() status.Status {
- workers := s.Workers()
+ workers := s.workers()
for i := 0; i < len(workers); i++ {
// If state of the worker is ready (at least 1)
// we assume, that plugin's worker pool is ready
diff --git a/plugins/informer/interface.go b/plugins/informer/interface.go
index 8e3b922b..45f44691 100644
--- a/plugins/informer/interface.go
+++ b/plugins/informer/interface.go
@@ -1,8 +1,10 @@
package informer
-import "github.com/spiral/roadrunner/v2/pkg/worker"
+import (
+ "github.com/spiral/roadrunner/v2/pkg/process"
+)
// Informer used to get workers from particular plugin or set of plugins
type Informer interface {
- Workers() []worker.BaseProcess
+ Workers() []process.State
}
diff --git a/plugins/informer/plugin.go b/plugins/informer/plugin.go
index 416c0112..98081d34 100644
--- a/plugins/informer/plugin.go
+++ b/plugins/informer/plugin.go
@@ -3,7 +3,7 @@ package informer
import (
endure "github.com/spiral/endure/pkg/container"
"github.com/spiral/errors"
- "github.com/spiral/roadrunner/v2/pkg/worker"
+ "github.com/spiral/roadrunner/v2/pkg/process"
"github.com/spiral/roadrunner/v2/plugins/logger"
)
@@ -21,7 +21,7 @@ func (p *Plugin) Init(log logger.Logger) error {
}
// Workers provides BaseProcess slice with workers for the requested plugin
-func (p *Plugin) Workers(name string) ([]worker.BaseProcess, error) {
+func (p *Plugin) Workers(name string) ([]process.State, error) {
const op = errors.Op("informer_plugin_workers")
svc, ok := p.registry[name]
if !ok {
@@ -49,7 +49,7 @@ func (p *Plugin) Name() string {
return PluginName
}
-// RPCService returns associated rpc service.
+// RPC returns associated rpc service.
func (p *Plugin) RPC() interface{} {
return &rpc{srv: p, log: p.log}
}
diff --git a/plugins/informer/rpc.go b/plugins/informer/rpc.go
index 55a9832b..b0b5b1af 100644
--- a/plugins/informer/rpc.go
+++ b/plugins/informer/rpc.go
@@ -1,9 +1,8 @@
package informer
import (
- "github.com/spiral/roadrunner/v2/pkg/worker"
+ "github.com/spiral/roadrunner/v2/pkg/process"
"github.com/spiral/roadrunner/v2/plugins/logger"
- "github.com/spiral/roadrunner/v2/tools"
)
type rpc struct {
@@ -14,7 +13,7 @@ type rpc struct {
// WorkerList contains list of workers.
type WorkerList struct {
// Workers is list of workers.
- Workers []tools.ProcessState `json:"workers"`
+ Workers []process.State `json:"workers"`
}
// List all resettable services.
@@ -38,15 +37,9 @@ func (rpc *rpc) Workers(service string, list *WorkerList) error {
return err
}
- list.Workers = make([]tools.ProcessState, 0)
- for _, w := range workers {
- ps, err := tools.WorkerProcessState(w.(worker.BaseProcess))
- if err != nil {
- continue
- }
+ // write actual processes
+ list.Workers = workers
- list.Workers = append(list.Workers, ps)
- }
rpc.log.Debug("list of workers", "workers", list.Workers)
rpc.log.Debug("successfully finished Workers method")
return nil
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..871c8f76
--- /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"`
+ RemainAfterExit bool `mapstructure:"remain_after_exit"`
+ RestartSec uint64 `mapstructure:"restart_sec"`
+}
+
+// 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.RestartSec == 0 {
+ val := c.Services[k]
+ val.RestartSec = 30
+ c.Services[k] = val
+ }
+ }
+ }
+}
diff --git a/plugins/service/plugin.go b/plugins/service/plugin.go
new file mode 100644
index 00000000..b5608ff2
--- /dev/null
+++ b/plugins/service/plugin.go
@@ -0,0 +1,106 @@
+package service
+
+import (
+ "sync"
+
+ "github.com/spiral/errors"
+ "github.com/spiral/roadrunner/v2/pkg/process"
+ "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 (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() {
+ // lock here, because Stop command might be invoked during the Serve
+ service.Lock()
+ defer service.Unlock()
+
+ 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, NewServiceProcess(
+ service.cfg.Services[k].RemainAfterExit,
+ service.cfg.Services[k].ExecTimeout,
+ service.cfg.Services[k].RestartSec,
+ service.cfg.Services[k].Command,
+ service.logger,
+ errCh,
+ ))
+ }
+ }
+
+ // start all processes
+ for i := 0; i < len(service.processes); i++ {
+ service.processes[i].start()
+ }
+ }()
+
+ return errCh
+}
+
+func (service *Plugin) Workers() []process.State {
+ service.Lock()
+ defer service.Unlock()
+ states := make([]process.State, 0, len(service.processes))
+ for i := 0; i < len(service.processes); i++ {
+ st, err := process.GeneralProcessState(service.processes[i].Pid, service.processes[i].rawCmd)
+ if err != nil {
+ continue
+ }
+ states = append(states, st)
+ }
+ return states
+}
+
+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..49219eb0
--- /dev/null
+++ b/plugins/service/process.go
@@ -0,0 +1,153 @@
+package service
+
+import (
+ "os/exec"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "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 from the plugin
+ rawCmd string
+ Pid int
+
+ // root plugin error chan
+ errCh chan error
+ // logger
+ log logger.Logger
+
+ ExecTimeout time.Duration
+ RemainAfterExit bool
+ RestartSec uint64
+
+ // process start time
+ startTime time.Time
+ stopped uint64
+}
+
+// NewServiceProcess constructs service process structure
+func NewServiceProcess(restartAfterExit bool, execTimeout time.Duration, restartDelay uint64, command string, l logger.Logger, errCh chan error) *Process {
+ return &Process{
+ rawCmd: command,
+ RestartSec: restartDelay,
+ ExecTimeout: execTimeout,
+ RemainAfterExit: restartAfterExit,
+ errCh: errCh,
+ log: l,
+ }
+}
+
+// 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")
+
+ // crate fat-process here
+ p.createProcess()
+
+ // non blocking process start
+ err := p.command.Start()
+ if err != nil {
+ p.errCh <- errors.E(op, err)
+ return
+ }
+
+ // start process waiting routine
+ go p.wait()
+ // execHandler checks for the execTimeout
+ go p.execHandler()
+ // save start time
+ p.startTime = time.Now()
+ p.Pid = p.command.Process.Pid
+}
+
+// create command for the process
+func (p *Process) createProcess() {
+ // cmdArgs contain command arguments if the command in form of: php <command> or ls <command> -i -b
+ 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
+ }
+ // redirect stderr into the Write function of the process.go
+ p.command.Stderr = p
+}
+
+// wait process for exit
+func (p *Process) wait() {
+ // Wait error doesn't matter here
+ err := p.command.Wait()
+ if err != nil {
+ p.log.Error("process wait error", "error", err)
+ }
+ // wait for restart delay
+ if p.RemainAfterExit {
+ // wait for the delay
+ time.Sleep(time.Second * time.Duration(p.RestartSec))
+ // and start command again
+ p.start()
+ }
+}
+
+// stop can be only sent by the Endure when plugin stopped
+func (p *Process) stop() {
+ atomic.StoreUint64(&p.stopped, 1)
+}
+
+func (p *Process) execHandler() {
+ tt := time.NewTicker(time.Second)
+ for range tt.C {
+ // lock here, because p.startTime could be changed during the check
+ p.Lock()
+ // if the exec timeout is set
+ if p.ExecTimeout != 0 {
+ // if stopped -> kill the process (SIGINT-> SIGKILL) and exit
+ if atomic.CompareAndSwapUint64(&p.stopped, 1, 1) {
+ err := p.command.Process.Signal(syscall.SIGINT)
+ if err != nil {
+ _ = p.command.Process.Signal(syscall.SIGKILL)
+ }
+ tt.Stop()
+ p.Unlock()
+ return
+ }
+
+ // 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()
+ tt.Stop()
+ return
+ }
+ }
+ p.Unlock()
+ }
+}
+
+// unsafe and fast []byte to string convert
+//go:inline
+func toString(data []byte) string {
+ return *(*string)(unsafe.Pointer(&data))
+}
diff --git a/plugins/static/plugin.go b/plugins/static/plugin.go
index 5f108701..76cb9e68 100644
--- a/plugins/static/plugin.go
+++ b/plugins/static/plugin.go
@@ -57,7 +57,7 @@ func (s *Plugin) Name() string {
return PluginName
}
-// middleware must return true if request/response pair is handled within the middleware.
+// Middleware must return true if request/response pair is handled within the middleware.
func (s *Plugin) Middleware(next http.Handler) http.HandlerFunc {
// Define the http.HandlerFunc
return func(w http.ResponseWriter, r *http.Request) {
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-23 21:14:35 +0000