Update get_task to return the queue of the task

[svasudevprasad]

get_task has been updated to return the queue from which the task was originally obtained. This is requried so that in case the task cannot be currently executed, it is added back to the right queue.

Currently observed that a progression task for a old device like flame from jobs-android queue was being pushed to
jobs-android-mte-pixel8 queue

[jonathanmetzman]

Why is add_task unable to add the task back to the correct queue?

[svasudevprasad]

This is because the task does not store in the queue from which it was obtained. So add task will add it to the bots default queue. With subqueues, this is not correct since the task could be picked from the generic android queue in case subqueue was empty. So without this check, the task from generic android queue gets pushed to the bot based subqueue causing issues.

[jonathanmetzman]

What do you think of adding an:

if android:
  get_queue_from_job()

here:

clusterfuzz/src/clusterfuzz/_internal/base/tasks/__init__.py

Line 672 in 4578334

[svasudevprasad]

That wont work due to

clusterfuzz/src/clusterfuzz/_internal/base/tasks/__init__.py

Line 363 in 4578334

task = get_regular_task(default_android_queue())

get_queue_from_job() will return the bot/job's subqueue

[jonathanmetzman]

Why wouldn't it get the correct queue? The task is added to the correct queue by some mechanism in the first place, can we use that?

[svasudevprasad]

Let me check

[jonathanmetzman]

A better option is to put the queue in the Task object. How invasive would this change be? It also would let us avoid the android special case.

[svasudevprasad]

I felt that was too invasive and hence went with this approach. Also queue information inside task did not make sense. Isint task supposed to be independent of queue ?

[jonathanmetzman]

I felt that was too invasive and hence went with this approach.

The current approach seems to force us to change a lot of functions for to deal with the subqueue problem.

Also queue information inside task did not make sense. Isint task supposed to be independent of queue ?

I think it is not independent in the subqeueue approach, and passing it alongside task doesn't really maintain this.

Could we use https://cloud.google.com/pubsub/docs/subscription-message-filter to avoid getting messages for the wrong platform/subplatform? I think it's worth looking to if deeper changes can solve the problems subqueue has at once. Maybe we can discuss this in the sync today.

[jonathanmetzman]

lgtm

[jonathanmetzman]

Just fix the tests please

[jonathanmetzman]

Here's failure:
Finished Step #4
Step #4: Ran 634 tests (3 skipped, 0 errors, 4 failures).
Step #4:
Step #4: FAILED (failures=4)
Step #4:
Step #4: Ran 172 tests in 47.548s
Step #4: ----------------------------------------------------------------------
Step #4:
Step #4:
Step #4: + test normal job jobs-linux
Step #4: - test normal job
Step #4: AssertionError: 'test normal job' != 'test normal job jobs-linux'
Step #4: self.assertEqual('test normal job', task.payload())
Step #4: File "/workspace/src/clusterfuzz/_internal/tests/appengine/common/tasks_test.py", line 101, in test_regular
Step #4: Traceback (most recent call last):
Step #4: ----------------------------------------------------------------------
Step #4: Test regular tasks.
Step #4: FAIL: test_regular (tests.appengine.common.tasks_test.with_cloud_emulators..decorator..Wrapped.test_regular)
Step #4: ======================================================================
Step #4:
Step #4:
Step #4: + test high job high-end-jobs-linux
Step #4: - test high job
Step #4: AssertionError: 'test high job' != 'test high job high-end-jobs-linux'
Step #4: self.assertEqual('test high job', task.payload())
Step #4: File "/workspace/src/clusterfuzz/_internal/tests/appengine/common/tasks_test.py", line 88, in test_high_end
Step #4: Traceback (most recent call last):
Step #4: ----------------------------------------------------------------------
Step #4: Test high end tasks.
Step #4: FAIL: test_high_end (tests.appengine.common.tasks_test.with_cloud_emulators..decorator..Wrapped.test_high_end)
Step #4: ======================================================================
Step #4:
Step #4:
Step #4: ? +++++++++++
Step #4: + test normal4 job jobs-linux
Step #4: - test normal4 job
Step #4: AssertionError: 'test normal4 job' != 'test normal4 job jobs-linux'
Step #4: self.assertEqual('test normal4 job', task.payload())
Step #4: File "/workspace/src/clusterfuzz/_internal/tests/appengine/common/tasks_test.py", line 127, in test_defer
Step #4: Traceback (most recent call last):
Step #4: ----------------------------------------------------------------------
Step #4: Test deferring tasks which shouldn't be run yet.
Step #4: FAIL: test_defer (tests.appengine.common.tasks_test.with_cloud_emulators..decorator..Wrapped.test_defer)
Step #4: ======================================================================
Step #4:
Step #4:
Step #4: ? +++++
Step #4: + test override job None
Step #4: - test override job
Step #4: AssertionError: 'test override job' != 'test override job None'
Step #4: self.assertEqual('test override job', task.payload())
Step #4: File "/workspace/src/clusterfuzz/_internal/tests/appengine/common/tasks_test.py", line 145, in test_command_override
Step #4: Traceback (most recent call last):
Step #4: ----------------------------------------------------------------------
Step #4: Test command override.
Step #4: FAIL: test_command_override (tests.appengine.common.tasks_test.with_cloud_emulators..decorator..Wrapped.test_command_override)