summaryrefslogtreecommitdiff
path: root/plugins
diff options
context:
space:
mode:
Diffstat (limited to 'plugins')
-rw-r--r--plugins/jobs/drivers/sqs/consumer.go8
-rw-r--r--plugins/jobs/drivers/sqs/item.go30
-rw-r--r--plugins/jobs/drivers/sqs/listener.go6
-rw-r--r--plugins/jobs/plugin.go15
-rw-r--r--plugins/jobs/rpc.go14
5 files changed, 44 insertions, 29 deletions
diff --git a/plugins/jobs/drivers/sqs/consumer.go b/plugins/jobs/drivers/sqs/consumer.go
index 5d741358..9ce37543 100644
--- a/plugins/jobs/drivers/sqs/consumer.go
+++ b/plugins/jobs/drivers/sqs/consumer.go
@@ -262,9 +262,11 @@ func (j *JobConsumer) Push(ctx context.Context, jb *job.Job) error {
}
func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
- // The new value for the message's visibility timeout (in seconds). Values range: 0
- // to 43200. Maximum: 12 hours.
- _, err := j.client.SendMessage(ctx, msg.pack(j.queueURL))
+ d, err := msg.pack(j.queueURL)
+ if err != nil {
+ return err
+ }
+ _, err = j.client.SendMessage(ctx, d)
if err != nil {
return err
}
diff --git a/plugins/jobs/drivers/sqs/item.go b/plugins/jobs/drivers/sqs/item.go
index f5fac0b3..eac06731 100644
--- a/plugins/jobs/drivers/sqs/item.go
+++ b/plugins/jobs/drivers/sqs/item.go
@@ -17,6 +17,7 @@ import (
const (
StringType string = "String"
NumberType string = "Number"
+ BinaryType string = "Binary"
ApproximateReceiveCount string = "ApproximateReceiveCount"
)
@@ -25,6 +26,7 @@ var itemAttributes = []string{
job.RRDelay,
job.RRTimeout,
job.RRPriority,
+ job.RRHeaders,
}
type Item struct {
@@ -128,7 +130,13 @@ func (i *Item) Ack() error {
}
func (i *Item) Nack() error {
- _, err := i.Options.client.DeleteMessage(context.Background(), &sqs.DeleteMessageInput{
+ // requeue message
+ err := i.Options.requeueFn(context.Background(), i)
+ if err != nil {
+ return err
+ }
+
+ _, err = i.Options.client.DeleteMessage(context.Background(), &sqs.DeleteMessageInput{
QueueUrl: i.Options.queue,
ReceiptHandle: i.Options.receiptHandler,
})
@@ -183,7 +191,13 @@ func fromJob(job *job.Job) *Item {
}
}
-func (i *Item) pack(queue *string) *sqs.SendMessageInput {
+func (i *Item) pack(queue *string) (*sqs.SendMessageInput, error) {
+ // pack headers map
+ data, err := json.Marshal(i.Headers)
+ if err != nil {
+ return nil, err
+ }
+
return &sqs.SendMessageInput{
MessageBody: aws.String(i.Payload),
QueueUrl: queue,
@@ -192,9 +206,10 @@ func (i *Item) pack(queue *string) *sqs.SendMessageInput {
job.RRJob: {DataType: aws.String(StringType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(i.Job)},
job.RRDelay: {DataType: aws.String(StringType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(strconv.Itoa(int(i.Options.Delay)))},
job.RRTimeout: {DataType: aws.String(StringType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(strconv.Itoa(int(i.Options.Timeout)))},
+ job.RRHeaders: {DataType: aws.String(BinaryType), BinaryValue: data, BinaryListValues: nil, StringListValues: nil, StringValue: nil},
job.RRPriority: {DataType: aws.String(NumberType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(strconv.Itoa(int(i.Options.Priority)))},
},
- }
+ }, nil
}
func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
@@ -210,6 +225,12 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
}
}
+ var h map[string][]string
+ err := json.Unmarshal(msg.MessageAttributes[job.RRHeaders].BinaryValue, &h)
+ if err != nil {
+ return nil, err
+ }
+
delay, err := strconv.Atoi(*msg.MessageAttributes[job.RRDelay].StringValue)
if err != nil {
return nil, errors.E(op, err)
@@ -233,6 +254,7 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
item := &Item{
Job: *msg.MessageAttributes[job.RRJob].StringValue,
Payload: *msg.Body,
+ Headers: h,
Options: &Options{
Delay: int64(delay),
Timeout: int64(to),
@@ -241,7 +263,7 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
// private
approxReceiveCount: int64(recCount),
client: j.client,
- queue: j.queue,
+ queue: j.queueURL,
receiptHandler: msg.ReceiptHandle,
requeueFn: j.handleItem,
},
diff --git a/plugins/jobs/drivers/sqs/listener.go b/plugins/jobs/drivers/sqs/listener.go
index a283b4a3..9efef90d 100644
--- a/plugins/jobs/drivers/sqs/listener.go
+++ b/plugins/jobs/drivers/sqs/listener.go
@@ -30,8 +30,10 @@ func (j *JobConsumer) listen(ctx context.Context) { //nolint:gocognit
MaxNumberOfMessages: j.prefetch,
AttributeNames: []types.QueueAttributeName{types.QueueAttributeName(ApproximateReceiveCount)},
MessageAttributeNames: []string{All},
- VisibilityTimeout: j.visibilityTimeout,
- WaitTimeSeconds: j.waitTime,
+ // The new value for the message's visibility timeout (in seconds). Values range: 0
+ // to 43200. Maximum: 12 hours.
+ VisibilityTimeout: j.visibilityTimeout,
+ WaitTimeSeconds: j.waitTime,
})
if err != nil {
diff --git a/plugins/jobs/plugin.go b/plugins/jobs/plugin.go
index 87559034..e2fffda7 100644
--- a/plugins/jobs/plugin.go
+++ b/plugins/jobs/plugin.go
@@ -30,17 +30,18 @@ const (
)
type Plugin struct {
- cfg *Config `structure:"jobs"`
- log logger.Logger
-
sync.RWMutex
+ // Jobs plugin configuration
+ cfg *Config `structure:"jobs"`
+ log logger.Logger
workersPool pool.Pool
server server.Server
jobConstructors map[string]jobs.Constructor
consumers map[string]jobs.Consumer
+ // events handler
events events.Handler
// priority queue implementation
@@ -55,6 +56,7 @@ type Plugin struct {
// signal channel to stop the pollers
stopCh chan struct{}
+ // internal payloads pool
pldPool sync.Pool
}
@@ -189,7 +191,7 @@ func (p *Plugin) Serve() chan error { //nolint:gocognit
jb := p.queue.ExtractMin()
// parse the context
- // for the each job, context contains:
+ // for each job, context contains:
/*
1. Job class
2. Job ID provided from the outside
@@ -216,12 +218,13 @@ func (p *Plugin) Serve() chan error { //nolint:gocognit
resp, err := p.workersPool.Exec(exec)
p.RUnlock()
if err != nil {
+ // RR protocol level error, Nack the job
errNack := jb.Nack()
if errNack != nil {
p.log.Error("negatively acknowledge failed", "error", errNack)
}
- p.log.Error("job execute", "error", err)
+ p.log.Error("job execute failed", "error", err)
p.putPayload(exec)
continue
@@ -310,7 +313,7 @@ func (p *Plugin) Reset() error {
defer p.Unlock()
const op = errors.Op("jobs_plugin_reset")
- p.log.Info("JOBS plugin got restart request. Restarting...")
+ p.log.Info("JOBS plugin received restart request. Restarting...")
p.workersPool.Destroy(context.Background())
p.workersPool = nil
diff --git a/plugins/jobs/rpc.go b/plugins/jobs/rpc.go
index 717ce33b..af1e12c0 100644
--- a/plugins/jobs/rpc.go
+++ b/plugins/jobs/rpc.go
@@ -13,20 +13,6 @@ type rpc struct {
p *Plugin
}
-/*
-List of the RPC methods:
-1. Release - single job push
-2. PushBatch - push job batch
-
-3. Reset - managed by the Resetter plugin
-
-4. Pause - pauses set of pipelines
-5. Resume - resumes set of pipelines
-
-6. Workers - managed by the Informer plugin.
-7. Stat - jobs statistic
-*/
-
func (r *rpc) Push(j *jobsv1beta.PushRequest, _ *jobsv1beta.Empty) error {
const op = errors.Op("jobs_rpc_push")
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-25 01:09:52 +0000