From 43071e43a0743ff8c7913bba7819952962124355 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 25 Jan 2021 22:47:02 +0300 Subject: Initial commit of the Temporal plugins set --- plugins/temporal/workflow/plugin.go | 198 ++++++++++++++++++++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 plugins/temporal/workflow/plugin.go (limited to 'plugins/temporal/workflow/plugin.go') diff --git a/plugins/temporal/workflow/plugin.go b/plugins/temporal/workflow/plugin.go new file mode 100644 index 00000000..ee2d722c --- /dev/null +++ b/plugins/temporal/workflow/plugin.go @@ -0,0 +1,198 @@ +package workflow + +import ( + "context" + "sync" + "sync/atomic" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/pkg/events" + "github.com/spiral/roadrunner/v2/pkg/worker" + "github.com/spiral/roadrunner/v2/plugins/logger" + "github.com/spiral/roadrunner/v2/plugins/server" + "github.com/spiral/roadrunner/v2/plugins/temporal/client" +) + +const ( + // PluginName defines public service name. + PluginName = "workflows" + + // RRMode sets as RR_MODE env variable to let worker know about the mode to run. + RRMode = "temporal/workflow" +) + +// Plugin manages workflows and workers. +type Plugin struct { + temporal client.Temporal + events events.Handler + server server.Server + log logger.Logger + mu sync.Mutex + reset chan struct{} + pool workflowPool + closing int64 +} + +// Init workflow plugin. +func (p *Plugin) Init(temporal client.Temporal, server server.Server, log logger.Logger) error { + p.temporal = temporal + p.server = server + p.events = events.NewEventsHandler() + p.log = log + p.reset = make(chan struct{}) + + return nil +} + +// Serve starts workflow service. +func (p *Plugin) Serve() chan error { + errCh := make(chan error, 1) + + pool, err := p.startPool() + if err != nil { + errCh <- errors.E("startPool", err) + return errCh + } + + p.pool = pool + + go func() { + for { + select { + case <-p.reset: + if atomic.LoadInt64(&p.closing) == 1 { + return + } + + err := p.replacePool() + if err == nil { + continue + } + + bkoff := backoff.NewExponentialBackOff() + bkoff.InitialInterval = time.Second + + err = backoff.Retry(p.replacePool, bkoff) + if err != nil { + errCh <- errors.E("deadPool", err) + } + } + } + }() + + return errCh +} + +// Stop workflow service. +func (p *Plugin) Stop() error { + atomic.StoreInt64(&p.closing, 1) + + pool := p.getPool() + if pool != nil { + p.pool = nil + return pool.Destroy(context.Background()) + } + + return nil +} + +// Name of the service. +func (p *Plugin) Name() string { + return PluginName +} + +// Workers returns list of available workflow workers. +func (p *Plugin) Workers() []worker.BaseProcess { + return p.pool.Workers() +} + +// WorkflowNames returns list of all available workflows. +func (p *Plugin) WorkflowNames() []string { + return p.pool.WorkflowNames() +} + +// Reset resets underlying workflow pool with new copy. +func (p *Plugin) Reset() error { + p.reset <- struct{}{} + + return nil +} + +// AddListener adds event listeners to the service. +func (p *Plugin) AddListener(listener events.Listener) { + p.events.AddListener(listener) +} + +// AddListener adds event listeners to the service. +func (p *Plugin) poolListener(event interface{}) { + if ev, ok := event.(PoolEvent); ok { + if ev.Event == eventWorkerExit { + if ev.Caused != nil { + p.log.Error("Workflow pool error", "error", ev.Caused) + } + p.reset <- struct{}{} + } + } + + p.events.Push(event) +} + +// AddListener adds event listeners to the service. +func (p *Plugin) startPool() (workflowPool, error) { + pool, err := newWorkflowPool( + p.temporal.GetCodec().WithLogger(p.log), + p.poolListener, + p.server, + ) + if err != nil { + return nil, errors.E(errors.Op("initWorkflowPool"), err) + } + + err = pool.Start(context.Background(), p.temporal) + if err != nil { + return nil, errors.E(errors.Op("startWorkflowPool"), err) + } + + p.log.Debug("Started workflow processing", "workflows", pool.WorkflowNames()) + + return pool, nil +} + +func (p *Plugin) replacePool() error { + p.mu.Lock() + defer p.mu.Unlock() + + if p.pool != nil { + errD := p.pool.Destroy(context.Background()) + p.pool = nil + if errD != nil { + p.log.Error( + "Unable to destroy expired workflow pool", + "error", + errors.E(errors.Op("destroyWorkflowPool"), errD), + ) + } + } + + pool, err := p.startPool() + if err != nil { + p.log.Error("Replace workflow pool failed", "error", err) + return errors.E(errors.Op("newWorkflowPool"), err) + } + + p.log.Debug("Replace workflow pool") + + p.pool = pool + + return nil +} + +// getPool returns currently pool. +func (p *Plugin) getPool() workflowPool { + p.mu.Lock() + defer p.mu.Unlock() + + return p.pool +} -- cgit v1.2.3 From 4638bdca80f75bc120b330022086d31c8b41be5b Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Tue, 26 Jan 2021 01:06:16 +0300 Subject: Code cleanup --- plugins/temporal/workflow/plugin.go | 37 ++++++++++++++++++++----------------- 1 file changed, 20 insertions(+), 17 deletions(-) (limited to 'plugins/temporal/workflow/plugin.go') diff --git a/plugins/temporal/workflow/plugin.go b/plugins/temporal/workflow/plugin.go index ee2d722c..3a397364 100644 --- a/plugins/temporal/workflow/plugin.go +++ b/plugins/temporal/workflow/plugin.go @@ -41,18 +41,19 @@ func (p *Plugin) Init(temporal client.Temporal, server server.Server, log logger p.server = server p.events = events.NewEventsHandler() p.log = log - p.reset = make(chan struct{}) + p.reset = make(chan struct{}, 1) return nil } // Serve starts workflow service. func (p *Plugin) Serve() chan error { + const op = errors.Op("workflow_plugin_serve") errCh := make(chan error, 1) pool, err := p.startPool() if err != nil { - errCh <- errors.E("startPool", err) + errCh <- errors.E(op, err) return errCh } @@ -76,7 +77,7 @@ func (p *Plugin) Serve() chan error { err = backoff.Retry(p.replacePool, bkoff) if err != nil { - errCh <- errors.E("deadPool", err) + errCh <- errors.E(op, err) } } } @@ -87,12 +88,17 @@ func (p *Plugin) Serve() chan error { // Stop workflow service. func (p *Plugin) Stop() error { + const op = errors.Op("workflow_plugin_stop") atomic.StoreInt64(&p.closing, 1) pool := p.getPool() if pool != nil { p.pool = nil - return pool.Destroy(context.Background()) + err := pool.Destroy(context.Background()) + if err != nil { + return errors.E(op, err) + } + return nil } return nil @@ -120,11 +126,6 @@ func (p *Plugin) Reset() error { return nil } -// AddListener adds event listeners to the service. -func (p *Plugin) AddListener(listener events.Listener) { - p.events.AddListener(listener) -} - // AddListener adds event listeners to the service. func (p *Plugin) poolListener(event interface{}) { if ev, ok := event.(PoolEvent); ok { @@ -141,18 +142,19 @@ func (p *Plugin) poolListener(event interface{}) { // AddListener adds event listeners to the service. func (p *Plugin) startPool() (workflowPool, error) { + const op = errors.Op("workflow_plugin_start_pool") pool, err := newWorkflowPool( p.temporal.GetCodec().WithLogger(p.log), p.poolListener, p.server, ) if err != nil { - return nil, errors.E(errors.Op("initWorkflowPool"), err) + return nil, errors.E(op, err) } err = pool.Start(context.Background(), p.temporal) if err != nil { - return nil, errors.E(errors.Op("startWorkflowPool"), err) + return nil, errors.E(op, err) } p.log.Debug("Started workflow processing", "workflows", pool.WorkflowNames()) @@ -162,29 +164,30 @@ func (p *Plugin) startPool() (workflowPool, error) { func (p *Plugin) replacePool() error { p.mu.Lock() + const op = errors.Op("workflow_plugin_replace_pool") defer p.mu.Unlock() if p.pool != nil { - errD := p.pool.Destroy(context.Background()) + err := p.pool.Destroy(context.Background()) p.pool = nil - if errD != nil { + if err != nil { p.log.Error( "Unable to destroy expired workflow pool", "error", - errors.E(errors.Op("destroyWorkflowPool"), errD), + errors.E(op, err), ) + return errors.E(op, err) } } pool, err := p.startPool() if err != nil { p.log.Error("Replace workflow pool failed", "error", err) - return errors.E(errors.Op("newWorkflowPool"), err) + return errors.E(op, err) } - p.log.Debug("Replace workflow pool") - p.pool = pool + p.log.Debug("workflow pool successfully replaced") return nil } -- cgit v1.2.3 From 800042d81fc9224ab05f01d1506df0d4b7ac4daa Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Tue, 26 Jan 2021 10:27:13 +0300 Subject: Attempt to fix data race in the Test_WorkerError_DisasterRecovery_Heavy test --- plugins/temporal/workflow/plugin.go | 2 ++ 1 file changed, 2 insertions(+) (limited to 'plugins/temporal/workflow/plugin.go') diff --git a/plugins/temporal/workflow/plugin.go b/plugins/temporal/workflow/plugin.go index 3a397364..572d9a3b 100644 --- a/plugins/temporal/workflow/plugin.go +++ b/plugins/temporal/workflow/plugin.go @@ -111,6 +111,8 @@ func (p *Plugin) Name() string { // Workers returns list of available workflow workers. func (p *Plugin) Workers() []worker.BaseProcess { + p.mu.Lock() + defer p.mu.Unlock() return p.pool.Workers() } -- cgit v1.2.3