Remove unneeded options, complete tests for the ephemeral, update proto

Signed-off-by: Valery Piashchynski <[email protected]>

7 files changed, 33 insertions, 128 deletions

diff --git a/plugins/jobs/drivers/amqp/item.go b/plugins/jobs/drivers/amqp/item.go
index 9b9625b0..5990d137 100644
--- a/plugins/jobs/drivers/amqp/item.go
+++ b/plugins/jobs/drivers/amqp/item.go
@@ -41,9 +41,6 @@ type Options struct {
 	// Delay defines time duration to delay execution for. Defaults to none.
 	Delay int64 `json:"delay,omitempty"`
 
-	// Reserve defines for how broker should wait until treating job are failed. Defaults to 30 min.
-	Timeout int64 `json:"timeout,omitempty"`
-
 	// private
 	// Ack delegates an acknowledgement through the Acknowledger interface that the client or server has finished work on a delivery
 	ack func(multiply bool) error
@@ -66,15 +63,6 @@ func (o *Options) DelayDuration() time.Duration {
 	return time.Second * time.Duration(o.Delay)
 }
 
-// TimeoutDuration returns timeout duration in a form of time.Duration.
-func (o *Options) TimeoutDuration() time.Duration {
-	if o.Timeout == 0 {
-		return 30 * time.Minute
-	}
-
-	return time.Second * time.Duration(o.Timeout)
-}
-
 func (i *Item) ID() string {
 	return i.Ident
 }
@@ -96,9 +84,8 @@ func (i *Item) Context() ([]byte, error) {
 			ID       string              `json:"id"`
 			Job      string              `json:"job"`
 			Headers  map[string][]string `json:"headers"`
-			Timeout  int64               `json:"timeout"`
 			Pipeline string              `json:"pipeline"`
-		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Timeout: i.Options.Timeout, Pipeline: i.Options.Pipeline},
+		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Pipeline: i.Options.Pipeline},
 	)
 
 	if err != nil {
@@ -141,10 +128,6 @@ func (i *Item) Requeue(headers map[string][]string, delay int64) error {
 	return nil
 }
 
-func (i *Item) Recycle() {
-	i.Options = nil
-}
-
 // fromDelivery converts amqp.Delivery into an Item which will be pushed to the PQ
 func (j *JobConsumer) fromDelivery(d amqp.Delivery) (*Item, error) {
 	const op = errors.Op("from_delivery_convert")
@@ -179,7 +162,6 @@ func fromJob(job *job.Job) *Item {
 			Priority: job.Options.Priority,
 			Pipeline: job.Options.Pipeline,
 			Delay:    job.Options.Delay,
-			Timeout:  job.Options.Timeout,
 		},
 	}
 }
@@ -195,7 +177,6 @@ func pack(id string, j *Item) (amqp.Table, error) {
 		job.RRJob:      j.Job,
 		job.RRPipeline: j.Options.Pipeline,
 		job.RRHeaders:  headers,
-		job.RRTimeout:  j.Options.Timeout,
 		job.RRDelay:    j.Options.Delay,
 		job.RRPriority: j.Options.Priority,
 	}, nil
@@ -232,10 +213,6 @@ func (j *JobConsumer) unpack(d amqp.Delivery) (*Item, error) {
 		}
 	}
 
-	if _, ok := d.Headers[job.RRTimeout].(int64); ok {
-		item.Options.Timeout = d.Headers[job.RRTimeout].(int64)
-	}
-
 	if _, ok := d.Headers[job.RRDelay].(int64); ok {
 		item.Options.Delay = d.Headers[job.RRDelay].(int64)
 	}
diff --git a/plugins/jobs/drivers/beanstalk/consumer.go b/plugins/jobs/drivers/beanstalk/consumer.go
index f41a2c8a..eaf99be1 100644
--- a/plugins/jobs/drivers/beanstalk/consumer.go
+++ b/plugins/jobs/drivers/beanstalk/consumer.go
@@ -201,7 +201,7 @@ func (j *JobConsumer) handleItem(ctx context.Context, item *Item) error {
 	// <ttr> seconds, the job will time out and the server will release the job.
 	//	The minimum ttr is 1. If the client sends 0, the server will silently
 	// increase the ttr to 1. Maximum ttr is 2**32-1.
-	id, err := j.pool.Put(ctx, bb.Bytes(), *j.tubePriority, item.Options.DelayDuration(), item.Options.TimeoutDuration())
+	id, err := j.pool.Put(ctx, bb.Bytes(), *j.tubePriority, item.Options.DelayDuration(), j.tout)
 	if err != nil {
 		errD := j.pool.Delete(ctx, id)
 		if errD != nil {
diff --git a/plugins/jobs/drivers/beanstalk/encode_test.go b/plugins/jobs/drivers/beanstalk/encode_test.go
index 34f2342b..e43207eb 100644
--- a/plugins/jobs/drivers/beanstalk/encode_test.go
+++ b/plugins/jobs/drivers/beanstalk/encode_test.go
@@ -26,7 +26,6 @@ func BenchmarkEncodeGob(b *testing.B) {
 			Priority: 10,
 			Pipeline: "test-local-pipe",
 			Delay:    10,
-			Timeout:  5,
 		},
 	}
 
@@ -60,7 +59,6 @@ func BenchmarkEncodeJsonIter(b *testing.B) {
 			Priority: 10,
 			Pipeline: "test-local-pipe",
 			Delay:    10,
-			Timeout:  5,
 		},
 	}
 
diff --git a/plugins/jobs/drivers/beanstalk/item.go b/plugins/jobs/drivers/beanstalk/item.go
index 47336b43..f1d7ac76 100644
--- a/plugins/jobs/drivers/beanstalk/item.go
+++ b/plugins/jobs/drivers/beanstalk/item.go
@@ -41,15 +41,6 @@ type Options struct {
 	// Delay defines time duration to delay execution for. Defaults to none.
 	Delay int64 `json:"delay,omitempty"`
 
-	// Reserve defines for how broker should wait until treating job are failed.
-	// - <ttr> -- time to run -- is an integer number of seconds to allow a worker
-	// to run this job. This time is counted from the moment a worker reserves
-	// this job. If the worker does not delete, release, or bury the job within
-	// <ttr> seconds, the job will time out and the server will release the job.
-	// The minimum ttr is 1. If the client sends 0, the server will silently
-	// increase the ttr to 1. Maximum ttr is 2**32-1.
-	Timeout int64 `json:"timeout,omitempty"`
-
 	// Private ================
 	id        uint64
 	conn      *beanstalk.Conn
@@ -61,15 +52,6 @@ func (o *Options) DelayDuration() time.Duration {
 	return time.Second * time.Duration(o.Delay)
 }
 
-// TimeoutDuration returns timeout duration in a form of time.Duration.
-func (o *Options) TimeoutDuration() time.Duration {
-	if o.Timeout == 0 {
-		return 30 * time.Minute
-	}
-
-	return time.Second * time.Duration(o.Timeout)
-}
-
 func (i *Item) ID() string {
 	return i.Ident
 }
@@ -91,9 +73,8 @@ func (i *Item) Context() ([]byte, error) {
 			ID       string              `json:"id"`
 			Job      string              `json:"job"`
 			Headers  map[string][]string `json:"headers"`
-			Timeout  int64               `json:"timeout"`
 			Pipeline string              `json:"pipeline"`
-		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Timeout: i.Options.Timeout, Pipeline: i.Options.Pipeline},
+		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Pipeline: i.Options.Pipeline},
 	)
 
 	if err != nil {
@@ -130,10 +111,6 @@ func (i *Item) Requeue(headers map[string][]string, delay int64) error {
 	return nil
 }
 
-func (i *Item) Recycle() {
-	i.Options = nil
-}
-
 func fromJob(job *job.Job) *Item {
 	return &Item{
 		Job:     job.Job,
@@ -144,7 +121,6 @@ func fromJob(job *job.Job) *Item {
 			Priority: job.Options.Priority,
 			Pipeline: job.Options.Pipeline,
 			Delay:    job.Options.Delay,
-			Timeout:  job.Options.Timeout,
 		},
 	}
 }
diff --git a/plugins/jobs/drivers/ephemeral/consumer.go b/plugins/jobs/drivers/ephemeral/consumer.go
index 03959b49..95ad6ecd 100644
--- a/plugins/jobs/drivers/ephemeral/consumer.go
+++ b/plugins/jobs/drivers/ephemeral/consumer.go
@@ -16,7 +16,8 @@ import (
 )
 
 const (
-	prefetch string = "prefetch"
+	prefetch      string = "prefetch"
+	goroutinesMax uint64 = 1000
 )
 
 type Config struct {
@@ -32,7 +33,7 @@ type JobConsumer struct {
 	localPrefetch chan *Item
 
 	// time.sleep goroutines max number
-	goroutinesMaxNum uint64
+	goroutines uint64
 
 	stopCh chan struct{}
 }
@@ -41,11 +42,11 @@ func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh
 	const op = errors.Op("new_ephemeral_pipeline")
 
 	jb := &JobConsumer{
-		log:              log,
-		pq:               pq,
-		eh:               eh,
-		goroutinesMaxNum: 1000,
-		stopCh:           make(chan struct{}, 1),
+		log:        log,
+		pq:         pq,
+		eh:         eh,
+		goroutines: 0,
+		stopCh:     make(chan struct{}, 1),
 	}
 
 	err := cfg.UnmarshalKey(configKey, &jb.cfg)
@@ -68,11 +69,11 @@ func NewJobBroker(configKey string, log logger.Logger, cfg config.Configurer, eh
 
 func FromPipeline(pipeline *pipeline.Pipeline, log logger.Logger, eh events.Handler, pq priorityqueue.Queue) (*JobConsumer, error) {
 	jb := &JobConsumer{
-		log:              log,
-		pq:               pq,
-		eh:               eh,
-		goroutinesMaxNum: 1000,
-		stopCh:           make(chan struct{}, 1),
+		log:        log,
+		pq:         pq,
+		eh:         eh,
+		goroutines: 0,
+		stopCh:     make(chan struct{}, 1),
 	}
 
 	// initialize a local queue
@@ -112,18 +113,18 @@ func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
 	// goroutines here. We should limit goroutines here.
 	if msg.Options.Delay > 0 {
 		// if we have 1000 goroutines waiting on the delay - reject 1001
-		if atomic.LoadUint64(&j.goroutinesMaxNum) >= 1000 {
+		if atomic.LoadUint64(&j.goroutines) >= goroutinesMax {
 			return errors.E(op, errors.Str("max concurrency number reached"))
 		}
 
 		go func(jj *Item) {
-			atomic.AddUint64(&j.goroutinesMaxNum, 1)
+			atomic.AddUint64(&j.goroutines, 1)
 			time.Sleep(jj.Options.DelayDuration())
 
 			// send the item after timeout expired
 			j.localPrefetch <- jj
 
-			atomic.AddUint64(&j.goroutinesMaxNum, ^uint64(0))
+			atomic.AddUint64(&j.goroutines, ^uint64(0))
 		}(msg)
 
 		return nil
@@ -149,7 +150,8 @@ func (j *JobConsumer) consume() {
 			}
 
 			// set requeue channel
-			item.Options.requeueCh = j.localPrefetch
+			item.Options.requeueFn = j.handleItem
+
 			j.pq.Insert(item)
 		case <-j.stopCh:
 			return
diff --git a/plugins/jobs/drivers/ephemeral/item.go b/plugins/jobs/drivers/ephemeral/item.go
index 9fab8d24..1a61d7e9 100644
--- a/plugins/jobs/drivers/ephemeral/item.go
+++ b/plugins/jobs/drivers/ephemeral/item.go
@@ -1,6 +1,7 @@
 package ephemeral
 
 import (
+	"context"
 	"time"
 
 	json "github.com/json-iterator/go"
@@ -37,11 +38,8 @@ type Options struct {
 	// Delay defines time duration to delay execution for. Defaults to none.
 	Delay int64 `json:"delay,omitempty"`
 
-	// Timeout defines for how broker should wait until treating job are failed. Defaults to 30 min.
-	Timeout int64 `json:"timeout,omitempty"`
-
 	// private
-	requeueCh chan *Item
+	requeueFn func(context.Context, *Item) error
 }
 
 // DelayDuration returns delay duration in a form of time.Duration.
@@ -49,15 +47,6 @@ func (o *Options) DelayDuration() time.Duration {
 	return time.Second * time.Duration(o.Delay)
 }
 
-// TimeoutDuration returns timeout duration in a form of time.Duration.
-func (o *Options) TimeoutDuration() time.Duration {
-	if o.Timeout == 0 {
-		return 30 * time.Minute
-	}
-
-	return time.Second * time.Duration(o.Timeout)
-}
-
 func (i *Item) ID() string {
 	return i.Ident
 }
@@ -78,9 +67,8 @@ func (i *Item) Context() ([]byte, error) {
 			ID       string              `json:"id"`
 			Job      string              `json:"job"`
 			Headers  map[string][]string `json:"headers"`
-			Timeout  int64               `json:"timeout"`
 			Pipeline string              `json:"pipeline"`
-		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Timeout: i.Options.Timeout, Pipeline: i.Options.Pipeline},
+		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Pipeline: i.Options.Pipeline},
 	)
 
 	if err != nil {
@@ -101,25 +89,16 @@ func (i *Item) Nack() error {
 }
 
 func (i *Item) Requeue(headers map[string][]string, delay int64) error {
-	go func() {
-		time.Sleep(time.Second * time.Duration(delay))
-		// overwrite the delay
-		i.Options.Delay = delay
-		i.Headers = headers
-		select {
-		case i.Options.requeueCh <- i:
-			return
-		default:
-			// TODO(rustatian): logs?
-			return
-		}
-	}()
+	// overwrite the delay
+	i.Options.Delay = delay
+	i.Headers = headers
 
-	return nil
-}
+	err := i.Options.requeueFn(context.Background(), i)
+	if err != nil {
+		return err
+	}
 
-func (i *Item) Recycle() {
-	i.Options = nil
+	return nil
 }
 
 func fromJob(job *job.Job) *Item {
@@ -131,7 +110,6 @@ func fromJob(job *job.Job) *Item {
 			Priority: job.Options.Priority,
 			Pipeline: job.Options.Pipeline,
 			Delay:    job.Options.Delay,
-			Timeout:  job.Options.Timeout,
 		},
 	}
 }
diff --git a/plugins/jobs/drivers/sqs/item.go b/plugins/jobs/drivers/sqs/item.go
index eac06731..df72b2e5 100644
--- a/plugins/jobs/drivers/sqs/item.go
+++ b/plugins/jobs/drivers/sqs/item.go
@@ -24,7 +24,6 @@ const (
 var itemAttributes = []string{
 	job.RRJob,
 	job.RRDelay,
-	job.RRTimeout,
 	job.RRPriority,
 	job.RRHeaders,
 }
@@ -58,9 +57,6 @@ type Options struct {
 	// Delay defines time duration to delay execution for. Defaults to none.
 	Delay int64 `json:"delay,omitempty"`
 
-	// Reserve defines for how broker should wait until treating job are failed. Defaults to 30 min.
-	Timeout int64 `json:"timeout,omitempty"`
-
 	// Private ================
 	approxReceiveCount int64
 	queue              *string
@@ -74,15 +70,6 @@ func (o *Options) DelayDuration() time.Duration {
 	return time.Second * time.Duration(o.Delay)
 }
 
-// TimeoutDuration returns timeout duration in a form of time.Duration.
-func (o *Options) TimeoutDuration() time.Duration {
-	if o.Timeout == 0 {
-		return 30 * time.Minute
-	}
-
-	return time.Second * time.Duration(o.Timeout)
-}
-
 func (i *Item) ID() string {
 	return i.Ident
 }
@@ -104,9 +91,8 @@ func (i *Item) Context() ([]byte, error) {
 			ID       string              `json:"id"`
 			Job      string              `json:"job"`
 			Headers  map[string][]string `json:"headers"`
-			Timeout  int64               `json:"timeout"`
 			Pipeline string              `json:"pipeline"`
-		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Timeout: i.Options.Timeout, Pipeline: i.Options.Pipeline},
+		}{ID: i.Ident, Job: i.Job, Headers: i.Headers, Pipeline: i.Options.Pipeline},
 	)
 
 	if err != nil {
@@ -172,10 +158,6 @@ func (i *Item) Requeue(headers map[string][]string, delay int64) error {
 	return nil
 }
 
-func (i *Item) Recycle() {
-	i.Options = nil
-}
-
 func fromJob(job *job.Job) *Item {
 	return &Item{
 		Job:     job.Job,
@@ -186,7 +168,6 @@ func fromJob(job *job.Job) *Item {
 			Priority: job.Options.Priority,
 			Pipeline: job.Options.Pipeline,
 			Delay:    job.Options.Delay,
-			Timeout:  job.Options.Timeout,
 		},
 	}
 }
@@ -205,7 +186,6 @@ func (i *Item) pack(queue *string) (*sqs.SendMessageInput, error) {
 		MessageAttributes: map[string]types.MessageAttributeValue{
 			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)))},
 		},
@@ -236,11 +216,6 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
 		return nil, errors.E(op, err)
 	}
 
-	to, err := strconv.Atoi(*msg.MessageAttributes[job.RRTimeout].StringValue)
-	if err != nil {
-		return nil, errors.E(op, err)
-	}
-
 	priority, err := strconv.Atoi(*msg.MessageAttributes[job.RRPriority].StringValue)
 	if err != nil {
 		return nil, errors.E(op, err)
@@ -257,7 +232,6 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
 		Headers: h,
 		Options: &Options{
 			Delay:    int64(delay),
-			Timeout:  int64(to),
 			Priority: int64(priority),
 
 			// private