summaryrefslogtreecommitdiff
path: root/plugins/jobs/brokers
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/jobs/brokers')
-rw-r--r--plugins/jobs/brokers/amqp/consumer.go67
-rw-r--r--plugins/jobs/brokers/amqp/plugin.go9
-rw-r--r--plugins/jobs/brokers/amqp/rabbit_init.go10
-rw-r--r--plugins/jobs/brokers/amqp/redial.go22
-rw-r--r--plugins/jobs/brokers/ephemeral/consumer.go53
-rw-r--r--plugins/jobs/brokers/ephemeral/plugin.go9
6 files changed, 108 insertions, 62 deletions
diff --git a/plugins/jobs/brokers/amqp/consumer.go b/plugins/jobs/brokers/amqp/consumer.go
index a7916f7e..481e102a 100644
--- a/plugins/jobs/brokers/amqp/consumer.go
+++ b/plugins/jobs/brokers/amqp/consumer.go
@@ -3,10 +3,12 @@ package amqp
import (
"fmt"
"sync"
+ "sync/atomic"
"time"
"github.com/google/uuid"
"github.com/spiral/errors"
+ "github.com/spiral/roadrunner/v2/pkg/events"
priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -48,8 +50,9 @@ type JobsConsumer struct {
sync.RWMutex
log logger.Logger
pq priorityqueue.Queue
+ eh events.Handler
- pipelines sync.Map
+ pipeline atomic.Value
// amqp connection
conn *amqp.Connection
@@ -71,7 +74,7 @@ type JobsConsumer struct {
}
// NewAMQPConsumer initializes rabbitmq pipeline
-func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer, pq priorityqueue.Queue) (*JobsConsumer, error) {
+func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobsConsumer, error) {
const op = errors.Op("new_amqp_consumer")
// we need to obtain two parts of the amqp information here.
// firs part - address to connect, it is located in the global section under the amqp pluginName
@@ -79,6 +82,7 @@ func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer,
jb := &JobsConsumer{
log: log,
pq: pq,
+ eh: e,
consumeID: uuid.NewString(),
stopCh: make(chan struct{}),
// TODO to config
@@ -146,13 +150,14 @@ func NewAMQPConsumer(configKey string, log logger.Logger, cfg config.Configurer,
return jb, nil
}
-func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Configurer, pq priorityqueue.Queue) (*JobsConsumer, error) {
+func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Configurer, e events.Handler, pq priorityqueue.Queue) (*JobsConsumer, error) {
const op = errors.Op("new_amqp_consumer_from_pipeline")
// we need to obtain two parts of the amqp information here.
// firs part - address to connect, it is located in the global section under the amqp pluginName
// second part - queues and other pipeline information
jb := &JobsConsumer{
log: log,
+ eh: e,
pq: pq,
consumeID: uuid.NewString(),
stopCh: make(chan struct{}),
@@ -210,8 +215,11 @@ func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, cfg config.Con
func (j *JobsConsumer) Push(job *structs.Job) error {
const op = errors.Op("rabbitmq_push")
// check if the pipeline registered
- if _, ok := j.pipelines.Load(job.Options.Pipeline); !ok {
- return errors.E(op, errors.Errorf("no such pipeline: %s", job.Options.Pipeline))
+
+ // load atomic value
+ pipe := j.pipeline.Load().(*pipeline.Pipeline)
+ if pipe.Name() != job.Options.Pipeline {
+ return errors.E(op, errors.Errorf("no such pipeline: %s, actual: %s", job.Options.Pipeline, pipe.Name()))
}
// lock needed here to protect redial concurrent operation
@@ -303,20 +311,16 @@ func (j *JobsConsumer) Push(job *structs.Job) error {
}
func (j *JobsConsumer) Register(pipeline *pipeline.Pipeline) error {
- const op = errors.Op("rabbitmq_register")
- if _, ok := j.pipelines.Load(pipeline.Name()); ok {
- return errors.E(op, errors.Errorf("queue %s has already been registered", pipeline))
- }
-
- j.pipelines.Store(pipeline.Name(), struct{}{})
-
+ j.pipeline.Store(pipeline)
return nil
}
-func (j *JobsConsumer) Run(pipeline *pipeline.Pipeline) error {
+func (j *JobsConsumer) Run(p *pipeline.Pipeline) error {
const op = errors.Op("rabbit_consume")
- if _, ok := j.pipelines.Load(pipeline.Name()); !ok {
- return errors.E(op, errors.Errorf("no such pipeline registered: %s", pipeline.Name()))
+
+ pipe := j.pipeline.Load().(*pipeline.Pipeline)
+ if pipe.Name() != p.Name() {
+ return errors.E(op, errors.Errorf("no such pipeline registered: %s", pipe.Name()))
}
// protect connection (redial)
@@ -354,22 +358,10 @@ func (j *JobsConsumer) Run(pipeline *pipeline.Pipeline) error {
return nil
}
-func (j *JobsConsumer) List() []string {
- out := make([]string, 0, 2)
-
- j.pipelines.Range(func(key, value interface{}) bool {
- pipe := key.(string)
- out = append(out, pipe)
- return true
- })
-
- return out
-}
-
-func (j *JobsConsumer) Pause(pipeline string) {
- if _, ok := j.pipelines.Load(pipeline); !ok {
- j.log.Error("no such pipeline", "requested pause on", pipeline)
- return
+func (j *JobsConsumer) Pause(p string) {
+ pipe := j.pipeline.Load().(*pipeline.Pipeline)
+ if pipe.Name() != p {
+ j.log.Error("no such pipeline", "requested pause on: ", p)
}
// protect connection (redial)
@@ -386,10 +378,10 @@ func (j *JobsConsumer) Pause(pipeline string) {
}
}
-func (j *JobsConsumer) Resume(pipeline string) {
- if _, ok := j.pipelines.Load(pipeline); !ok {
- j.log.Error("no such pipeline", "requested pause on", pipeline)
- return
+func (j *JobsConsumer) Resume(p string) {
+ pipe := j.pipeline.Load().(*pipeline.Pipeline)
+ if pipe.Name() != p {
+ j.log.Error("no such pipeline", "requested resume on: ", p)
}
// protect connection (redial)
@@ -430,11 +422,6 @@ func (j *JobsConsumer) Resume(pipeline string) {
func (j *JobsConsumer) Stop() error {
j.stopCh <- struct{}{}
- j.pipelines.Range(func(key, _ interface{}) bool {
- j.pipelines.Delete(key)
- return true
- })
-
return nil
}
diff --git a/plugins/jobs/brokers/amqp/plugin.go b/plugins/jobs/brokers/amqp/plugin.go
index ca972c5b..624f4405 100644
--- a/plugins/jobs/brokers/amqp/plugin.go
+++ b/plugins/jobs/brokers/amqp/plugin.go
@@ -2,6 +2,7 @@ package amqp
import (
"github.com/spiral/roadrunner/v2/common/jobs"
+ "github.com/spiral/roadrunner/v2/pkg/events"
priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -29,11 +30,11 @@ func (p *Plugin) Name() string {
func (p *Plugin) Available() {}
-func (p *Plugin) JobsConstruct(configKey string, pq priorityqueue.Queue) (jobs.Consumer, error) {
- return NewAMQPConsumer(configKey, p.log, p.cfg, pq)
+func (p *Plugin) JobsConstruct(configKey string, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error) {
+ return NewAMQPConsumer(configKey, p.log, p.cfg, e, pq)
}
// FromPipeline constructs AMQP driver from pipeline
-func (p *Plugin) FromPipeline(pipe *pipeline.Pipeline, pq priorityqueue.Queue) (jobs.Consumer, error) {
- return FromPipeline(pipe, p.log, p.cfg, pq)
+func (p *Plugin) FromPipeline(pipe *pipeline.Pipeline, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error) {
+ return FromPipeline(pipe, p.log, p.cfg, e, pq)
}
diff --git a/plugins/jobs/brokers/amqp/rabbit_init.go b/plugins/jobs/brokers/amqp/rabbit_init.go
index cb9f2dc4..e3e5f8da 100644
--- a/plugins/jobs/brokers/amqp/rabbit_init.go
+++ b/plugins/jobs/brokers/amqp/rabbit_init.go
@@ -1,11 +1,14 @@
package amqp
import (
+ "time"
+
"github.com/spiral/errors"
+ "github.com/spiral/roadrunner/v2/pkg/events"
)
func (j *JobsConsumer) initRabbitMQ() error {
- const op = errors.Op("rabbit_initmq")
+ const op = errors.Op("jobs_plugin_rmq_init")
// Channel opens a unique, concurrent server channel to process the bulk of AMQP
// messages. Any error from methods on this receiver will render the receiver
// invalid and a new Channel should be opened.
@@ -53,5 +56,10 @@ func (j *JobsConsumer) initRabbitMQ() error {
return errors.E(op, err)
}
+ j.eh.Push(events.JobEvent{
+ Event: events.EventInitialized,
+ Driver: "amqp",
+ Start: time.Now(),
+ })
return channel.Close()
}
diff --git a/plugins/jobs/brokers/amqp/redial.go b/plugins/jobs/brokers/amqp/redial.go
index 277e75b7..571ee548 100644
--- a/plugins/jobs/brokers/amqp/redial.go
+++ b/plugins/jobs/brokers/amqp/redial.go
@@ -2,9 +2,12 @@ package amqp
import (
"fmt"
+ "time"
"github.com/cenkalti/backoff/v4"
"github.com/spiral/errors"
+ "github.com/spiral/roadrunner/v2/pkg/events"
+ "github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
"github.com/streadway/amqp"
)
@@ -22,6 +25,17 @@ func (j *JobsConsumer) redialer() { //nolint:gocognit
j.Lock()
+ t := time.Now()
+ pipe := j.pipeline.Load().(*pipeline.Pipeline)
+ j.eh.Push(events.JobEvent{
+ Event: events.EventPipeError,
+ Pipeline: pipe.Name(),
+ Driver: pipe.Driver(),
+ Error: err,
+ Start: time.Now(),
+ Elapsed: 0,
+ })
+
j.log.Error("connection closed, reconnecting", "error", err)
expb := backoff.NewExponentialBackOff()
// set the retry timeout (minutes)
@@ -85,6 +99,14 @@ func (j *JobsConsumer) redialer() { //nolint:gocognit
return
}
+ j.eh.Push(events.JobEvent{
+ Event: events.EventPipeActive,
+ Pipeline: pipe.Name(),
+ Driver: pipe.Driver(),
+ Start: t,
+ Elapsed: time.Since(t),
+ })
+
j.Unlock()
case <-j.stopCh:
diff --git a/plugins/jobs/brokers/ephemeral/consumer.go b/plugins/jobs/brokers/ephemeral/consumer.go
index b51af322..559cb2e9 100644
--- a/plugins/jobs/brokers/ephemeral/consumer.go
+++ b/plugins/jobs/brokers/ephemeral/consumer.go
@@ -5,6 +5,7 @@ import (
"time"
"github.com/spiral/errors"
+ "github.com/spiral/roadrunner/v2/pkg/events"
priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -23,19 +24,21 @@ type Config struct {
type JobBroker struct {
cfg *Config
log logger.Logger
- queues sync.Map
+ eh events.Handler
+ pipeline sync.Map
pq priorityqueue.Queue
localQueue chan *Item
stopCh chan struct{}
}
-func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, pq priorityqueue.Queue) (*JobBroker, error) {
+func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh events.Handler, pq priorityqueue.Queue) (*JobBroker, error) {
const op = errors.Op("new_ephemeral_pipeline")
jb := &JobBroker{
log: log,
pq: pq,
+ eh: eh,
stopCh: make(chan struct{}, 1),
}
@@ -57,10 +60,11 @@ func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, pq
return jb, nil
}
-func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, pq priorityqueue.Queue) (*JobBroker, error) {
+func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, eh events.Handler, pq priorityqueue.Queue) (*JobBroker, error) {
jb := &JobBroker{
log: log,
pq: pq,
+ eh: eh,
stopCh: make(chan struct{}, 1),
}
@@ -79,7 +83,7 @@ func (j *JobBroker) Push(job *structs.Job) error {
const op = errors.Op("ephemeral_push")
// check if the pipeline registered
- if b, ok := j.queues.Load(job.Options.Pipeline); ok {
+ if b, ok := j.pipeline.Load(job.Options.Pipeline); ok {
if !b.(bool) {
return errors.E(op, errors.Errorf("pipeline disabled: %s", job.Options.Pipeline))
}
@@ -119,37 +123,51 @@ func (j *JobBroker) consume() {
func (j *JobBroker) Register(pipeline *pipeline.Pipeline) error {
const op = errors.Op("ephemeral_register")
- if _, ok := j.queues.Load(pipeline.Name()); ok {
+ if _, ok := j.pipeline.Load(pipeline.Name()); ok {
return errors.E(op, errors.Errorf("queue %s has already been registered", pipeline))
}
- j.queues.Store(pipeline.Name(), true)
+ j.pipeline.Store(pipeline.Name(), true)
return nil
}
func (j *JobBroker) Pause(pipeline string) {
- if q, ok := j.queues.Load(pipeline); ok {
+ if q, ok := j.pipeline.Load(pipeline); ok {
if q == true {
// mark pipeline as turned off
- j.queues.Store(pipeline, false)
+ j.pipeline.Store(pipeline, false)
}
}
+
+ j.eh.Push(events.JobEvent{
+ Event: events.EventPipeStopped,
+ Pipeline: pipeline,
+ Start: time.Now(),
+ Elapsed: 0,
+ })
}
func (j *JobBroker) Resume(pipeline string) {
- if q, ok := j.queues.Load(pipeline); ok {
+ if q, ok := j.pipeline.Load(pipeline); ok {
if q == false {
// mark pipeline as turned off
- j.queues.Store(pipeline, true)
+ j.pipeline.Store(pipeline, true)
}
}
+
+ j.eh.Push(events.JobEvent{
+ Event: events.EventPipeActive,
+ Pipeline: pipeline,
+ Start: time.Now(),
+ Elapsed: 0,
+ })
}
func (j *JobBroker) List() []string {
out := make([]string, 0, 2)
- j.queues.Range(func(key, value interface{}) bool {
+ j.pipeline.Range(func(key, value interface{}) bool {
pipe := key.(string)
out = append(out, pipe)
return true
@@ -164,13 +182,22 @@ func (j *JobBroker) Run(_ *pipeline.Pipeline) error {
}
func (j *JobBroker) Stop() error {
- j.queues.Range(func(key, _ interface{}) bool {
- j.queues.Delete(key)
+ var pipe string
+ j.pipeline.Range(func(key, _ interface{}) bool {
+ pipe = key.(string)
+ j.pipeline.Delete(key)
return true
})
// return from the consumer
j.stopCh <- struct{}{}
+ j.eh.Push(events.JobEvent{
+ Event: events.EventPipeStopped,
+ Pipeline: pipe,
+ Start: time.Now(),
+ Elapsed: 0,
+ })
+
return nil
}
diff --git a/plugins/jobs/brokers/ephemeral/plugin.go b/plugins/jobs/brokers/ephemeral/plugin.go
index bfe2d6ac..28495abb 100644
--- a/plugins/jobs/brokers/ephemeral/plugin.go
+++ b/plugins/jobs/brokers/ephemeral/plugin.go
@@ -2,6 +2,7 @@ package ephemeral
import (
"github.com/spiral/roadrunner/v2/common/jobs"
+ "github.com/spiral/roadrunner/v2/pkg/events"
priorityqueue "github.com/spiral/roadrunner/v2/pkg/priority_queue"
"github.com/spiral/roadrunner/v2/plugins/config"
"github.com/spiral/roadrunner/v2/plugins/jobs/pipeline"
@@ -30,11 +31,11 @@ func (p *Plugin) Name() string {
func (p *Plugin) Available() {}
// JobsConstruct creates new ephemeral consumer from the configuration
-func (p *Plugin) JobsConstruct(configKey string, pq priorityqueue.Queue) (jobs.Consumer, error) {
- return NewJobBroker(configKey, p.log, p.cfg, pq)
+func (p *Plugin) JobsConstruct(configKey string, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error) {
+ return NewJobBroker(configKey, p.log, p.cfg, e, pq)
}
// FromPipeline creates new ephemeral consumer from the provided pipeline
-func (p *Plugin) FromPipeline(pipeline *pipeline.Pipeline, pq priorityqueue.Queue) (jobs.Consumer, error) {
- return FromPipeline(pipeline, p.log, pq)
+func (p *Plugin) FromPipeline(pipeline *pipeline.Pipeline, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error) {
+ return FromPipeline(pipeline, p.log, e, pq)
}