Fix various bugs in the SQS. Implement SQS tests for the

jobs_ok.php/jobs_err.php workers.

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

11 files changed, 254 insertions, 53 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")
 
diff --git a/tests/jobs_ok.php b/tests/jobs_ok.php
index fa58dd9a..4e786d15 100644
--- a/tests/jobs_ok.php
+++ b/tests/jobs_ok.php
@@ -19,12 +19,7 @@ while ($in = $rr->waitPayload()) {
 
         $rr->respond(new RoadRunner\Payload(json_encode([
             'type' => 0,
-            'data' => [
-                'message' => 'error',
-                'requeue' => true,
-                'delay_seconds' => 10,
-                'headers' => $headers
-            ]
+            'data' => []
         ])));
     } catch (\Throwable $e) {
         $rr->error((string)$e);
diff --git a/tests/plugins/jobs/jobs_sqs_test.go b/tests/plugins/jobs/jobs_sqs_test.go
index 7cad3876..d4cb4e52 100644
--- a/tests/plugins/jobs/jobs_sqs_test.go
+++ b/tests/plugins/jobs/jobs_sqs_test.go
@@ -24,6 +24,7 @@ import (
 	jobsv1beta "github.com/spiral/roadrunner/v2/proto/jobs/v1beta"
 	"github.com/spiral/roadrunner/v2/tests/mocks"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestSQSInit(t *testing.T) {
@@ -47,19 +48,21 @@ func TestSQSInit(t *testing.T) {
 	mockLogger.EXPECT().Info("pipeline started", "pipeline", "test-1", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
 	mockLogger.EXPECT().Info("pipeline started", "pipeline", "test-2", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
 
-	mockLogger.EXPECT().Info("driver initialized", "driver", "amqp", "start", gomock.Any()).Times(2)
+	mockLogger.EXPECT().Info("pipeline active", "pipeline", "test-1", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Info("pipeline active", "pipeline", "test-2", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+
+	mockLogger.EXPECT().Info("driver initialized", "driver", "sqs", "start", gomock.Any()).Times(2)
 
 	mockLogger.EXPECT().Warn("pipeline stopped", "pipeline", "test-1", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
 	mockLogger.EXPECT().Warn("pipeline stopped", "pipeline", "test-2", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
 
-	mockLogger.EXPECT().Info("delivery channel closed, leaving the rabbit listener").Times(2)
+	mockLogger.EXPECT().Warn("sqs listener stopped").Times(2)
 
 	err = cont.RegisterAll(
 		cfg,
 		&server.Plugin{},
 		&rpcPlugin.Plugin{},
-		&logger.ZapLogger{},
-		// mockLogger,
+		mockLogger,
 		&jobs.Plugin{},
 		&resetter.Plugin{},
 		&informer.Plugin{},
@@ -135,22 +138,113 @@ func TestSQSDeclare(t *testing.T) {
 	mockLogger.EXPECT().Debug("Started RPC service", "address", "tcp://127.0.0.1:6001", "services", gomock.Any()).Times(1)
 	mockLogger.EXPECT().Error(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 
-	mockLogger.EXPECT().Info("pipeline started", "pipeline", "test-1", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
-	mockLogger.EXPECT().Info("pipeline started", "pipeline", "test-2", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Info("pipeline active", "pipeline", "test-3", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Info("pipeline paused", "pipeline", "test-3", "driver", "sqs", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Warn("pipeline stopped", "pipeline", "test-3", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Warn("sqs listener stopped").Times(1)
 
-	mockLogger.EXPECT().Info("driver initialized", "driver", "amqp", "start", gomock.Any()).Times(2)
+	err = cont.RegisterAll(
+		cfg,
+		&server.Plugin{},
+		&rpcPlugin.Plugin{},
+		mockLogger,
+		&jobs.Plugin{},
+		&resetter.Plugin{},
+		&informer.Plugin{},
+		&sqs.Plugin{},
+	)
+	assert.NoError(t, err)
 
-	mockLogger.EXPECT().Warn("pipeline stopped", "pipeline", "test-1", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
-	mockLogger.EXPECT().Warn("pipeline stopped", "pipeline", "test-2", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	err = cont.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch, err := cont.Serve()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	stopCh := make(chan struct{}, 1)
 
-	mockLogger.EXPECT().Info("delivery channel closed, leaving the rabbit listener").Times(2)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case e := <-ch:
+				assert.Fail(t, "error", e.Error.Error())
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+			case <-sig:
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			case <-stopCh:
+				// timeout
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			}
+		}
+	}()
+
+	time.Sleep(time.Second * 3)
+
+	t.Run("DeclareSQSPipeline", declareSQSPipe)
+	t.Run("ConsumeSQSPipeline", resumePipes("test-3"))
+	t.Run("PushSQSPipeline", pushToPipe("test-3"))
+	time.Sleep(time.Second)
+	t.Run("PauseSQSPipeline", pausePipelines("test-3"))
+	time.Sleep(time.Second)
+	t.Run("DestroySQSPipeline", destroyPipelines("test-3"))
+
+	time.Sleep(time.Second * 5)
+	stopCh <- struct{}{}
+	wg.Wait()
+}
+
+func TestSQSJobsError(t *testing.T) {
+	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
+	assert.NoError(t, err)
+
+	cfg := &config.Viper{
+		Path:   "sqs/.rr-sqs-jobs-err.yaml",
+		Prefix: "rr",
+	}
+
+	controller := gomock.NewController(t)
+	mockLogger := mocks.NewMockLogger(controller)
+
+	// general
+	mockLogger.EXPECT().Debug("worker destructed", "pid", gomock.Any()).AnyTimes()
+	mockLogger.EXPECT().Debug("worker constructed", "pid", gomock.Any()).AnyTimes()
+	mockLogger.EXPECT().Debug("Started RPC service", "address", "tcp://127.0.0.1:6001", "services", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Error(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
+
+	mockLogger.EXPECT().Info("pipeline active", "pipeline", "test-3", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Error("jobs protocol error", "error", "error", "delay", gomock.Any(), "requeue", gomock.Any()).Times(3)
+
+	mockLogger.EXPECT().Info("pipeline paused", "pipeline", "test-3", "driver", "sqs", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Warn("pipeline stopped", "pipeline", "test-3", "start", gomock.Any(), "elapsed", gomock.Any()).Times(1)
+	mockLogger.EXPECT().Warn("sqs listener stopped").Times(1)
 
 	err = cont.RegisterAll(
 		cfg,
 		&server.Plugin{},
 		&rpcPlugin.Plugin{},
-		&logger.ZapLogger{},
-		// mockLogger,
+		mockLogger,
 		&jobs.Plugin{},
 		&resetter.Plugin{},
 		&informer.Plugin{},
@@ -208,12 +302,46 @@ func TestSQSDeclare(t *testing.T) {
 	t.Run("DeclareSQSPipeline", declareSQSPipe)
 	t.Run("ConsumeSQSPipeline", resumePipes("test-3"))
 	t.Run("PushSQSPipeline", pushToPipe("test-3"))
+	time.Sleep(time.Second * 25)
 	t.Run("PauseSQSPipeline", pausePipelines("test-3"))
+	time.Sleep(time.Second)
 	t.Run("DestroySQSPipeline", destroyPipelines("test-3"))
 
 	time.Sleep(time.Second * 5)
 	stopCh <- struct{}{}
 	wg.Wait()
+
+	time.Sleep(time.Second * 5)
+}
+
+func TestSQSNoGlobalSection(t *testing.T) {
+	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
+	assert.NoError(t, err)
+
+	cfg := &config.Viper{
+		Path:   "sqs/.rr-no-global.yaml",
+		Prefix: "rr",
+	}
+
+	err = cont.RegisterAll(
+		cfg,
+		&server.Plugin{},
+		&rpcPlugin.Plugin{},
+		&logger.ZapLogger{},
+		&jobs.Plugin{},
+		&resetter.Plugin{},
+		&informer.Plugin{},
+		&sqs.Plugin{},
+	)
+	assert.NoError(t, err)
+
+	err = cont.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = cont.Serve()
+	require.Error(t, err)
 }
 
 func declareSQSPipe(t *testing.T) {
diff --git a/tests/plugins/jobs/sqs/.rr-no-global.yaml b/tests/plugins/jobs/sqs/.rr-no-global.yaml
new file mode 100644
index 00000000..2c97a37e
--- /dev/null
+++ b/tests/plugins/jobs/sqs/.rr-no-global.yaml
@@ -0,0 +1,39 @@
+rpc:
+  listen: tcp://127.0.0.1:6001
+
+server:
+  command: "php ../../jobs_ok.php"
+  relay: "pipes"
+  relay_timeout: "20s"
+
+logs:
+  level: error
+  mode: development
+
+jobs:
+  num_pollers: 10
+  pipeline_size: 100000
+  pool:
+    num_workers: 10
+    max_jobs: 0
+    allocate_timeout: 60s
+    destroy_timeout: 60s
+
+  pipelines:
+    test-1:
+      driver: sqs
+      prefetch: 1000
+      visibility_timeout: 0
+      wait_time_seconds: 0
+      queue: default
+      attributes:
+        DelaySeconds: 0
+        MaximumMessageSize: 262144
+        MessageRetentionPeriod: 345600
+        ReceiveMessageWaitTimeSeconds: 0
+        VisibilityTimeout: 30
+      tags:
+        test: "tag"
+
+  consume: [ "test-1" ]
+
diff --git a/tests/plugins/jobs/sqs/.rr-sqs-declare.yaml b/tests/plugins/jobs/sqs/.rr-sqs-declare.yaml
index c4eb9cf9..21209cbb 100644
--- a/tests/plugins/jobs/sqs/.rr-sqs-declare.yaml
+++ b/tests/plugins/jobs/sqs/.rr-sqs-declare.yaml
@@ -2,7 +2,7 @@ rpc:
   listen: tcp://127.0.0.1:6001
 
 server:
-  command: "php ../../client.php echo pipes"
+  command: "php ../../jobs_ok.php"
   relay: "pipes"
   relay_timeout: "20s"
 
@@ -20,7 +20,7 @@ logs:
   mode: development
 
 jobs:
-  num_pollers: 10
+  num_pollers: 1
   pipeline_size: 100000
   pool:
     num_workers: 10
diff --git a/tests/plugins/jobs/sqs/.rr-sqs-init.yaml b/tests/plugins/jobs/sqs/.rr-sqs-init.yaml
index 239f9954..ffdec1fd 100644
--- a/tests/plugins/jobs/sqs/.rr-sqs-init.yaml
+++ b/tests/plugins/jobs/sqs/.rr-sqs-init.yaml
@@ -33,7 +33,6 @@ jobs:
       visibility_timeout: 0
       wait_time_seconds: 0
       queue: default
-      # https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SetQueueAttributes.html
       attributes:
         DelaySeconds: 0
         MaximumMessageSize: 262144
@@ -51,8 +50,5 @@ jobs:
         MessageRetentionPeriod: 86400
       tags:
         test: "tag"
-
-
-  # list of pipelines to be consumed by the server, keep empty if you want to start consuming manually
   consume: [ "test-1", "test-2" ]
 
diff --git a/tests/plugins/jobs/sqs/.rr-sqs-jobs-err.yaml b/tests/plugins/jobs/sqs/.rr-sqs-jobs-err.yaml
new file mode 100644
index 00000000..b518d433
--- /dev/null
+++ b/tests/plugins/jobs/sqs/.rr-sqs-jobs-err.yaml
@@ -0,0 +1,28 @@
+rpc:
+  listen: tcp://127.0.0.1:6001
+
+server:
+  command: "php ../../jobs_err.php"
+  relay: "pipes"
+  relay_timeout: "20s"
+
+sqs:
+  key: api-key
+  secret: api-secret
+  region: us-west-1
+  endpoint: http://127.0.0.1:9324
+
+logs:
+  level: debug
+  encoding: console
+  mode: development
+
+jobs:
+  num_pollers: 10
+  timeout: 60
+  pipeline_size: 100000
+  pool:
+    num_workers: 10
+    max_jobs: 0
+    allocate_timeout: 60s
+    destroy_timeout: 60s