diff options
author | Valery Piashchynski <[email protected]> | 2021-04-30 16:32:11 +0300 |
---|---|---|
committer | Valery Piashchynski <[email protected]> | 2021-04-30 16:32:11 +0300 |
commit | 4236dfac2cd65a9031b04aa659448152868a4190 (patch) | |
tree | 591e77440a7f595d595c72cf1129e235501a75b1 | |
parent | 38021decff98dabcab873f4c258f12d122988cdd (diff) | |
parent | 009b7009885d8a15e6fa6c7e78436087b2f20129 (diff) |
Merge branch 'beta' into stable
32 files changed, 155 insertions, 54 deletions
diff --git a/CHANGELOG.md b/CHANGELOG.md index f9291f32..26612551 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,12 +13,15 @@ v2.2.0 (11.05.2021) , `weak` and `pattern`. ### Option `always` was deleted from the plugin. -- ✏️ + +- ✏️ Update `informer.List` implementation. Now it returns a list with the all available plugins in the runtime. ## 🩹 Fixes: - 🐛 Fix: issue with wrong ordered middlewares (reverse). Now the order is correct. +--- + v2.1.1 (29.04.2021) ------------------- @@ -47,9 +47,6 @@ test_coverage: go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/redis.out -covermode=atomic ./tests/plugins/redis go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/resetter.out -covermode=atomic ./tests/plugins/resetter go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/rpc.out -covermode=atomic ./tests/plugins/rpc - go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/boltdb.out -covermode=atomic ./tests/plugins/kv/boltdb - go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/memory.out -covermode=atomic ./tests/plugins/kv/memory - go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/memcached.out -covermode=atomic ./tests/plugins/kv/memcached go test -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./coverage/kv_plugin.out -covermode=atomic ./tests/plugins/kv cat ./coverage/*.out > ./coverage/summary.out docker-compose -f tests/docker-compose.yaml down diff --git a/pkg/events/pool_events.go b/pkg/events/pool_events.go index c569a5a8..e7b451e0 100644 --- a/pkg/events/pool_events.go +++ b/pkg/events/pool_events.go @@ -1,6 +1,5 @@ package events -// TODO event numbers const ( // EventWorkerConstruct thrown when new worker is spawned. EventWorkerConstruct P = iota + 10000 diff --git a/pkg/pool/static_pool.go b/pkg/pool/static_pool.go index 129c6f7d..06005d98 100755 --- a/pkg/pool/static_pool.go +++ b/pkg/pool/static_pool.go @@ -153,7 +153,6 @@ func (sp *StaticPool) Exec(p payload.Payload) (payload.Payload, error) { } // worker want's to be terminated - // TODO careful with toString(rsp.Context) if len(rsp.Body) == 0 && toString(rsp.Context) == StopRequest { sp.stopWorker(w) return sp.Exec(p) diff --git a/pkg/transport/pipe/pipe_factory.go b/pkg/transport/pipe/pipe_factory.go index 0709413f..19f4f92d 100755 --- a/pkg/transport/pipe/pipe_factory.go +++ b/pkg/transport/pipe/pipe_factory.go @@ -42,7 +42,6 @@ func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd, lis return } - // TODO why out is in? in, err := cmd.StdoutPipe() if err != nil { c <- SpawnResult{ @@ -52,7 +51,6 @@ func (f *Factory) SpawnWorkerWithTimeout(ctx context.Context, cmd *exec.Cmd, lis return } - // TODO why in is out? out, err := cmd.StdinPipe() if err != nil { c <- SpawnResult{ @@ -119,13 +117,11 @@ func (f *Factory) SpawnWorker(cmd *exec.Cmd, listeners ...events.Listener) (*wor return nil, errors.E(op, err) } - // TODO why out is in? in, err := cmd.StdoutPipe() if err != nil { return nil, errors.E(op, err) } - // TODO why in is out? out, err := cmd.StdinPipe() if err != nil { return nil, errors.E(op, err) diff --git a/plugins/config/plugin.go b/plugins/config/plugin.go index 58647eb8..918381c4 100755 --- a/plugins/config/plugin.go +++ b/plugins/config/plugin.go @@ -10,6 +10,8 @@ import ( "github.com/spiral/errors" ) +const PluginName string = "config" + type Viper struct { viper *viper.Viper Path string @@ -113,11 +115,27 @@ func (v *Viper) Has(name string) bool { return v.viper.IsSet(name) } -// Returns common config parameters +// GetCommonConfig Returns common config parameters func (v *Viper) GetCommonConfig() *General { return v.CommonConfig } +func (v *Viper) Serve() chan error { + return make(chan error, 1) +} + +func (v *Viper) Stop() error { + return nil +} + +// Name returns user-friendly plugin name +func (v *Viper) Name() string { + return PluginName +} + +// Available interface implementation +func (v *Viper) Available() {} + func parseFlag(flag string) (string, string, error) { const op = errors.Op("parse_flag") if !strings.Contains(flag, "=") { diff --git a/plugins/gzip/plugin.go b/plugins/gzip/plugin.go index 18ee7b88..24b125fb 100644 --- a/plugins/gzip/plugin.go +++ b/plugins/gzip/plugin.go @@ -10,7 +10,7 @@ const PluginName = "gzip" type Plugin struct{} -// needed for the Endure +// Init needed for the Endure func (g *Plugin) Init() error { return nil } @@ -21,6 +21,9 @@ func (g *Plugin) Middleware(next http.Handler) http.Handler { }) } +// Available interface implementation +func (g *Plugin) Available() {} + func (g *Plugin) Name() string { return PluginName } diff --git a/plugins/headers/plugin.go b/plugins/headers/plugin.go index dea0d127..19c444df 100644 --- a/plugins/headers/plugin.go +++ b/plugins/headers/plugin.go @@ -8,11 +8,11 @@ import ( "github.com/spiral/roadrunner/v2/plugins/config" ) -// ID contains default service name. +// PluginName contains default service name. const PluginName = "headers" const RootPluginName = "http" -// Service serves headers files. Potentially convert into middleware? +// Plugin serves headers files. Potentially convert into middleware? type Plugin struct { // server configuration (location, forbidden files and etc) cfg *Config @@ -37,7 +37,7 @@ func (s *Plugin) Init(cfg config.Configurer) error { return nil } -// middleware must return true if request/response pair is handled within the middleware. +// Middleware is HTTP plugin middleware to serve headers func (s *Plugin) Middleware(next http.Handler) http.Handler { // Define the http.HandlerFunc return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -69,6 +69,9 @@ func (s *Plugin) Name() string { return PluginName } +// Available interface implementation +func (s *Plugin) Available() {} + // configure OPTIONS response func (s *Plugin) preflightRequest(w http.ResponseWriter) { headers := w.Header() diff --git a/plugins/http/plugin.go b/plugins/http/plugin.go index 2b1dec89..344102f4 100644 --- a/plugins/http/plugin.go +++ b/plugins/http/plugin.go @@ -440,3 +440,6 @@ func (s *Plugin) Ready() status.Status { Code: http.StatusServiceUnavailable, } } + +// Available interface implementation +func (s *Plugin) Available() {} diff --git a/plugins/informer/interface.go b/plugins/informer/interface.go index 45f44691..316c7bc1 100644 --- a/plugins/informer/interface.go +++ b/plugins/informer/interface.go @@ -4,7 +4,18 @@ import ( "github.com/spiral/roadrunner/v2/pkg/process" ) +/* +Informer plugin should not receive any other plugin in the Init or via Collects +Because Availabler implementation should present in every plugin +*/ + // Informer used to get workers from particular plugin or set of plugins type Informer interface { Workers() []process.State } + +// Availabler interface should be implemented by every plugin which wish to report to the PHP worker that it available in the RR runtime +type Availabler interface { + // Available method needed to collect all plugins which are available in the runtime. + Available() +} diff --git a/plugins/informer/plugin.go b/plugins/informer/plugin.go index 98081d34..2d76123b 100644 --- a/plugins/informer/plugin.go +++ b/plugins/informer/plugin.go @@ -4,19 +4,18 @@ import ( endure "github.com/spiral/endure/pkg/container" "github.com/spiral/errors" "github.com/spiral/roadrunner/v2/pkg/process" - "github.com/spiral/roadrunner/v2/plugins/logger" ) const PluginName = "informer" type Plugin struct { - registry map[string]Informer - log logger.Logger + registry map[string]Informer + available map[string]Availabler } -func (p *Plugin) Init(log logger.Logger) error { +func (p *Plugin) Init() error { + p.available = make(map[string]Availabler) p.registry = make(map[string]Informer) - p.log = log return nil } @@ -31,19 +30,24 @@ func (p *Plugin) Workers(name string) ([]process.State, error) { return svc.Workers(), nil } -// CollectTarget resettable service. -func (p *Plugin) CollectTarget(name endure.Named, r Informer) error { - p.registry[name.Name()] = r - return nil -} - // Collects declares services to be collected. func (p *Plugin) Collects() []interface{} { return []interface{}{ - p.CollectTarget, + p.CollectPlugins, + p.CollectWorkers, } } +// CollectPlugins collects all RR plugins +func (p *Plugin) CollectPlugins(name endure.Named, l Availabler) { + p.available[name.Name()] = l +} + +// CollectWorkers obtains plugins with workers inside. +func (p *Plugin) CollectWorkers(name endure.Named, r Informer) { + p.registry[name.Name()] = r +} + // Name of the service. func (p *Plugin) Name() string { return PluginName @@ -51,5 +55,5 @@ func (p *Plugin) Name() string { // RPC returns associated rpc service. func (p *Plugin) RPC() interface{} { - return &rpc{srv: p, log: p.log} + return &rpc{srv: p} } diff --git a/plugins/informer/rpc.go b/plugins/informer/rpc.go index b0b5b1af..8955af92 100644 --- a/plugins/informer/rpc.go +++ b/plugins/informer/rpc.go @@ -2,12 +2,10 @@ package informer import ( "github.com/spiral/roadrunner/v2/pkg/process" - "github.com/spiral/roadrunner/v2/plugins/logger" ) type rpc struct { srv *Plugin - log logger.Logger } // WorkerList contains list of workers. @@ -18,20 +16,17 @@ type WorkerList struct { // List all resettable services. func (rpc *rpc) List(_ bool, list *[]string) error { - rpc.log.Debug("Started List method") *list = make([]string, 0, len(rpc.srv.registry)) - for name := range rpc.srv.registry { + // append all plugin names to the output result + for name := range rpc.srv.available { *list = append(*list, name) } - rpc.log.Debug("list of services", "list", *list) - rpc.log.Debug("successfully finished List method") return nil } // Workers state of a given service. func (rpc *rpc) Workers(service string, list *WorkerList) error { - rpc.log.Debug("started Workers method", "service", service) workers, err := rpc.srv.Workers(service) if err != nil { return err @@ -40,7 +35,5 @@ func (rpc *rpc) Workers(service string, list *WorkerList) error { // write actual processes list.Workers = workers - rpc.log.Debug("list of workers", "workers", list.Workers) - rpc.log.Debug("successfully finished Workers method") return nil } diff --git a/plugins/kv/config.go b/plugins/kv/config.go index 9ecae644..66095817 100644 --- a/plugins/kv/config.go +++ b/plugins/kv/config.go @@ -1,5 +1,6 @@ package kv +// Config represents general storage configuration with keys as the user defined kv-names and values as the drivers type Config struct { Data map[string]interface{} `mapstructure:"kv"` } diff --git a/plugins/kv/drivers/boltdb/driver.go b/plugins/kv/drivers/boltdb/driver.go index b596d4c3..2e2df527 100644 --- a/plugins/kv/drivers/boltdb/driver.go +++ b/plugins/kv/drivers/boltdb/driver.go @@ -401,8 +401,6 @@ func (d *Driver) startGCLoop() { //nolint:gocognit }) if err != nil { d.log.Error("error during the gc phase of update", "error", err) - // todo this error is ignored, it means, that timer still be active - // to prevent this, we need to invoke t.Stop() return false } } diff --git a/plugins/kv/drivers/boltdb/plugin.go b/plugins/kv/drivers/boltdb/plugin.go index 9d1e0dba..9b4cf9f7 100644 --- a/plugins/kv/drivers/boltdb/plugin.go +++ b/plugins/kv/drivers/boltdb/plugin.go @@ -63,3 +63,6 @@ func (s *Plugin) Provide(key string) (kv.Storage, error) { func (s *Plugin) Name() string { return PluginName } + +// Available interface implementation +func (s *Plugin) Available() {} diff --git a/plugins/kv/drivers/memcached/plugin.go b/plugins/kv/drivers/memcached/plugin.go index af59e91b..cde84f42 100644 --- a/plugins/kv/drivers/memcached/plugin.go +++ b/plugins/kv/drivers/memcached/plugin.go @@ -33,6 +33,9 @@ func (s *Plugin) Name() string { return PluginName } +// Available interface implementation +func (s *Plugin) Available() {} + func (s *Plugin) Provide(key string) (kv.Storage, error) { const op = errors.Op("boltdb_plugin_provide") st, err := NewMemcachedDriver(s.log, key, s.cfgPlugin) diff --git a/plugins/kv/drivers/memory/plugin.go b/plugins/kv/drivers/memory/plugin.go index acc6023d..2be7caae 100644 --- a/plugins/kv/drivers/memory/plugin.go +++ b/plugins/kv/drivers/memory/plugin.go @@ -62,3 +62,7 @@ func (s *Plugin) Provide(key string) (kv.Storage, error) { func (s *Plugin) Name() string { return PluginName } + +// Available interface implementation +func (s *Plugin) Available() { +} diff --git a/plugins/kv/storage.go b/plugins/kv/storage.go index e8468b77..fe2fa10b 100644 --- a/plugins/kv/storage.go +++ b/plugins/kv/storage.go @@ -184,3 +184,7 @@ func (p *Plugin) RPC() interface{} { func (p *Plugin) Name() string { return PluginName } + +// Available interface implementation +func (p *Plugin) Available() { +} diff --git a/plugins/logger/config.go b/plugins/logger/config.go index c435e8be..d2236cac 100644 --- a/plugins/logger/config.go +++ b/plugins/logger/config.go @@ -40,7 +40,7 @@ type Config struct { ErrorOutput []string `mapstructure:"errorOutput"` } -// ZapConfig converts config into Zap configuration. +// BuildLogger converts config into Zap configuration. func (cfg *Config) BuildLogger() (*zap.Logger, error) { var zCfg zap.Config switch strings.ToLower(cfg.Mode) { @@ -49,7 +49,29 @@ func (cfg *Config) BuildLogger() (*zap.Logger, error) { case "production": zCfg = zap.NewProductionConfig() case "development": - zCfg = zap.NewDevelopmentConfig() + zCfg = zap.Config{ + Level: zap.NewAtomicLevelAt(zap.DebugLevel), + Development: true, + Encoding: "console", + EncoderConfig: zapcore.EncoderConfig{ + // Keys can be anything except the empty string. + TimeKey: "T", + LevelKey: "L", + NameKey: "N", + CallerKey: "C", + FunctionKey: zapcore.OmitKey, + MessageKey: "M", + StacktraceKey: "S", + LineEnding: zapcore.DefaultLineEnding, + EncodeLevel: ColoredLevelEncoder, + EncodeTime: zapcore.ISO8601TimeEncoder, + EncodeDuration: zapcore.StringDurationEncoder, + EncodeCaller: zapcore.ShortCallerEncoder, + EncodeName: ColoredNameEncoder, + }, + OutputPaths: []string{"stderr"}, + ErrorOutputPaths: []string{"stderr"}, + } case "raw": zCfg = zap.Config{ Level: zap.NewAtomicLevelAt(zap.InfoLevel), @@ -98,12 +120,10 @@ func (cfg *Config) BuildLogger() (*zap.Logger, error) { zCfg.ErrorOutputPaths = cfg.ErrorOutput } - // todo: - return zCfg.Build() } -// Initialize default logger +// InitDefault Initialize default logger func (cfg *Config) InitDefault() { if cfg.Mode == "" { cfg.Mode = "development" diff --git a/plugins/logger/plugin.go b/plugins/logger/plugin.go index 9965b29c..ffbf7f5e 100644 --- a/plugins/logger/plugin.go +++ b/plugins/logger/plugin.go @@ -75,3 +75,12 @@ func (z *ZapLogger) Provides() []interface{} { z.ServiceLogger, } } + +// Name returns user-friendly plugin name +func (z *ZapLogger) Name() string { + return PluginName +} + +// Available interface implementation +func (z *ZapLogger) Available() { +} diff --git a/plugins/metrics/plugin.go b/plugins/metrics/plugin.go index d0443177..474bb21d 100644 --- a/plugins/metrics/plugin.go +++ b/plugins/metrics/plugin.go @@ -222,3 +222,7 @@ func (m *Plugin) RPC() interface{} { log: m.log, } } + +// Available interface implementation +func (m *Plugin) Available() { +} diff --git a/plugins/redis/plugin.go b/plugins/redis/plugin.go index 204abd17..2eab7043 100644 --- a/plugins/redis/plugin.go +++ b/plugins/redis/plugin.go @@ -76,3 +76,7 @@ func (s Plugin) Stop() error { func (s *Plugin) Name() string { return PluginName } + +// Available interface implementation +func (s *Plugin) Available() { +} diff --git a/plugins/reload/plugin.go b/plugins/reload/plugin.go index bf88462e..7e6bdfec 100644 --- a/plugins/reload/plugin.go +++ b/plugins/reload/plugin.go @@ -161,3 +161,7 @@ func (s *Plugin) Stop() error { func (s *Plugin) Name() string { return PluginName } + +// Available interface implementation +func (s *Plugin) Available() { +} diff --git a/plugins/reload/watcher.go b/plugins/reload/watcher.go index 1b3407e5..421668b3 100644 --- a/plugins/reload/watcher.go +++ b/plugins/reload/watcher.go @@ -216,9 +216,8 @@ func (w *Watcher) waitEvent(d time.Duration) error { case <-ticker.C: // this is not very effective way // because we have to wait on Lock - // better is to listen files in parallel, but, since that would be used in debug... TODO + // better is to listen files in parallel, but, since that would be used in debug... for serviceName := range w.watcherConfigs { - // TODO sync approach fileList, _ := w.retrieveFileList(serviceName, w.watcherConfigs[serviceName]) w.pollEvents(w.watcherConfigs[serviceName].ServiceName, fileList) } diff --git a/plugins/resetter/plugin.go b/plugins/resetter/plugin.go index 43382e56..4feb692a 100644 --- a/plugins/resetter/plugin.go +++ b/plugins/resetter/plugin.go @@ -74,6 +74,10 @@ func (p *Plugin) Name() string { return PluginName } +// Available interface implementation +func (p *Plugin) Available() { +} + // RPC returns associated rpc service. func (p *Plugin) RPC() interface{} { return &rpc{srv: p, log: p.log} diff --git a/plugins/rpc/plugin.go b/plugins/rpc/plugin.go index b80994d3..cfe20321 100644 --- a/plugins/rpc/plugin.go +++ b/plugins/rpc/plugin.go @@ -122,6 +122,10 @@ func (s *Plugin) Name() string { return PluginName } +// Available interface implementation +func (s *Plugin) Available() { +} + // Collects all plugins which implement Name + RPCer interfaces func (s *Plugin) Collects() []interface{} { return []interface{}{ diff --git a/plugins/server/plugin.go b/plugins/server/plugin.go index c3496ae7..22b568d8 100644 --- a/plugins/server/plugin.go +++ b/plugins/server/plugin.go @@ -58,6 +58,10 @@ func (server *Plugin) Name() string { return PluginName } +// Available interface implementation +func (server *Plugin) Available() { +} + // Serve (Start) server plugin (just a mock here to satisfy interface) func (server *Plugin) Serve() chan error { const op = errors.Op("server_plugin_serve") diff --git a/plugins/service/plugin.go b/plugins/service/plugin.go index b5608ff2..28b84443 100644 --- a/plugins/service/plugin.go +++ b/plugins/service/plugin.go @@ -104,3 +104,7 @@ func (service *Plugin) Stop() error { func (service *Plugin) Name() string { return PluginName } + +// Available interface implementation +func (service *Plugin) Available() { +} diff --git a/plugins/status/plugin.go b/plugins/status/plugin.go index dc4e506d..82a0fa6c 100644 --- a/plugins/status/plugin.go +++ b/plugins/status/plugin.go @@ -127,7 +127,10 @@ func (c *Plugin) Name() string { return PluginName } -// RPCService returns associated rpc service. +// Available interface implementation +func (c *Plugin) Available() {} + +// RPC returns associated rpc service. func (c *Plugin) RPC() interface{} { return &rpc{srv: c, log: c.log} } diff --git a/tests/plugins/informer/informer_test.go b/tests/plugins/informer/informer_test.go index b6f50fd5..d40d7093 100644 --- a/tests/plugins/informer/informer_test.go +++ b/tests/plugins/informer/informer_test.go @@ -64,10 +64,7 @@ func TestInformerInit(t *testing.T) { 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 { @@ -113,9 +110,11 @@ func informerListRPCTest(t *testing.T) { assert.NoError(t, err) client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) // WorkerList contains list of workers. - list := make([]string, 0, 0) + list := make([]string, 0, 5) + // Plugins which are expected to be in the list + expected := []string{"rpc", "logs", "informer.plugin1", "config", "server"} err = client.Call("informer.List", true, &list) assert.NoError(t, err) - assert.Equal(t, "informer.plugin1", list[0]) + assert.ElementsMatch(t, list, expected) } diff --git a/tests/plugins/informer/test_plugin.go b/tests/plugins/informer/test_plugin.go index 0c9065a3..43335999 100644 --- a/tests/plugins/informer/test_plugin.go +++ b/tests/plugins/informer/test_plugin.go @@ -49,6 +49,8 @@ func (p1 *Plugin1) Name() string { return "informer.plugin1" } +func (p1 *Plugin1) Available() {} + func (p1 *Plugin1) Workers() []process.State { p, err := p1.server.NewWorkerPool(context.Background(), testPoolConfig, nil) if err != nil { diff --git a/tools/worker_table.go b/tools/worker_table.go index 4dd70d8e..799f045f 100644 --- a/tools/worker_table.go +++ b/tools/worker_table.go @@ -83,7 +83,6 @@ func renderStatus(status string) string { } func renderJobs(number uint64) string { - // TODO overflow return humanize.Comma(int64(number)) } |