[fix #1726] Use boto3 for SQS async requests #1759

rafidka · 2023-06-22T00:24:21Z

Overview

Context on this can be found in #1726 , but briefly, AWS SQS is planning on a change for making JSON the default communication protocol instead of XML for their boto3 library. Currently, kombu manually craft the AWS request, which results in an XML response when it hits SQS, yet uses boto3 to parse the response. So, when the latter changes to JSON, it will break Celery when used with SQS (and, consequently, break Airflow when used with Celery and SQS.)

This PR fixes the issue by using boto3 all the way, instead of just for parsing responses. Thus, when AWS SQS eventually changes the communication protocol for its boto3 client from XML to JSON, the requests will ask for JSON and the response will be parsed as JSON, thus avoiding the problem altogether.

Testing

To reproduce the issue, I created this simple Celery application:

worker.py

from celery import Celery

app = Celery('worker', broker='sqs://')

app.config_from_object
app.conf.update(
    broker_transport_options={
        'region': 'us-west-2',
        'visibility_timeout': 300,
        'polling_interval': 1,
        'predefined_queues': {
            'celery': {
                'url': 'https://sqs.us-west-2.amazonaws.com/<account ID>/celery-test',
            }
        }
    },
)


@app.task(acks_late=False)
def add(x, y):
    print(x + y)
    return x + y

scheduler.py

import time

from worker import add

while True:
    print("Scheduling task...")
    add.delay(4, 4)
    time.sleep(3)

Then I specifically installed botocore version 1.29.127, which switched to JSON formatting (before it was rolled back in the next version) and confirmed the issue is happening. After my fix in this PR, the issue disappeared.

auvipy

can you please check the build failures?

rafidka · 2023-06-22T03:09:54Z

can you please check the build failures?

Yeah, did notice the lint broke. I will fix it. Meanwhile, I would still appreciate feedback on the code. Thanks.

auvipy · 2023-06-22T04:38:25Z

it is not lint broke :) its a test failure =================================== FAILURES ===================================
_________________________ test_Channel.test_get_async __________________________

self = <t.unit.transport.test_SQS.test_Channel object at 0x7f7f958ff610>

@pytest.mark.usefixtures('hub')
def test_get_async(self):
    """Basic coverage of async code typically used via:
    basic_consume > _loop1 > _schedule_queue > _get_bulk_async"""
    # Prepare
    for i in range(3):
        message = 'message: %s' % i
        self.producer.publish(message)

    # SQS.Channel.asynsqs constructs AsyncSQSConnection using self.sqs
    # which is already a mock thanks to `setup` above, we just need to
    # mock the async-specific methods (as test_AsyncSQSConnection does)
    async_sqs_conn = self.channel.asynsqs(self.queue_name)
    async_sqs_conn.get_list = Mock(name='X.get_list')

    # Call key method
    self.channel._get_bulk_async(self.queue_name)

  assert async_sqs_conn.get_list.call_count == 1

E AssertionError: assert 0 == 1
E + where 0 = .call_count
E + where = <kombu.asynchronous.aws.sqs.connection.AsyncSQSConnection object at 0x7f7f82b4a090>.get_list

rafidka · 2023-06-22T22:21:51Z

it is not lint broke :) its a test failure

Ah, ok, sorry I missed that. I ran tests locally, but I had to scope it to some of the files only as I started getting some Azure related errors. @auvipy , what is the easiest way to run all tests locally?

rafidka · 2023-06-23T00:12:22Z

@auvipy , I fixed the build issues. Could you please take a look?

auvipy · 2023-06-23T03:06:51Z

yes, thanks! I am going to review it

auvipy

the codes look good. can you also check in the docs to see if anything need adjustment? thanks for your effort here

rafidka · 2023-06-23T05:13:35Z

the codes look good. can you also check in the docs to see if anything need adjustment? thanks for your effort here

Sure, will do.

for more information, see https://pre-commit.ci

rafidka · 2023-06-26T13:55:20Z

the codes look good. can you also check in the docs to see if anything need adjustment? thanks for your effort here

Hi @auvipy . I checked the documentation under the doc/ folder and I didn't see anything related to the methods I removed so we should be good. I also rebased my branch and fixed a couple of issues the linter was complaining about. Finally, 3 members from my team also reviewed the change and they think the change looks good. Please let me know if there is anything else that I can help with.

One important thing I would like to mention is regarding the methods I removed. I confirmed that those methods are not being used within Celery, but I cannot confirm they are not referenced outside. However, since those methods are not documented, and they are kind of internal methods, they are not supposed to be used by others, and if that's the case, then the using code should be updated accordingly, much like how we are updating kombu to avoid using botocore's internal stuff. Please let me your thoughts on this.

auvipy · 2023-06-27T03:51:58Z

I agree with your view points, I am going to review it and merge when feel confident. thanks for taking on this

auvipy · 2023-06-27T04:07:06Z

Thanks a lot Rafid!

auvipy · 2023-09-18T11:28:01Z

we got a report on performance regression here #1783 (comment). would be thankful to have this verified

This reverts commit 862d0bc.

auvipy · 2023-09-26T04:55:32Z

we might need to revert the change #1799

rafidka · 2023-09-26T21:21:58Z

@auvipy , I don't think this is a good idea. Reverting the change is a ticking bomb that will result in kombu being unusable completely when botocore is updated. Performance degradation seems the lesser of the two evils, assuming it is indeed caused by my PR. What is the timeline for the new release? I can try to find sometime to debug the issue to see if I can root cause the performance degradation issue.

auvipy · 2023-09-27T04:46:36Z

@auvipy , I don't think this is a good idea. Reverting the change is a ticking bomb that will result in kombu being unusable completely when botocore is updated. Performance degradation seems the lesser of the two evils, assuming it is indeed caused by my PR. What is the timeline for the new release? I can try to find sometime to debug the issue to see if I can root cause the performance degradation issue.

reverting is the last resort. you can take your time to re investigate the issue in coming weeks.

rafidka · 2023-09-27T16:09:14Z

@auvipy , I don't think this is a good idea. Reverting the change is a ticking bomb that will result in kombu being unusable completely when botocore is updated. Performance degradation seems the lesser of the two evils, assuming it is indeed caused by my PR. What is the timeline for the new release? I can try to find sometime to debug the issue to see if I can root cause the performance degradation issue.

reverting is the last resort. you can take your time to re investigate the issue in coming weeks.

@auvipy , what is the time frame we are talking about? I have an important deadline on October 18. If that's too far, I will try to find someone else in my team to take a look at this.

auvipy · 2023-09-27T16:24:19Z

That is a little bit too late. We want to cut a release this month, for bug fix stuff. But since it is a big change we can stop reverting for now and wait till 30 Oct, 2023 to find a Solution.

rafidka · 2023-09-29T19:57:00Z

We think that there indeed is an issue with Pull Request, and my colleague and I are actively working on root causing and fixing this issue. We will keep you updated.

rafidka · 2023-10-04T22:35:01Z

@auvipy , I found the root cause behind the performance degradation. Please see my comment here.

…)" This reverts commit 862d0bc.

TL;DR - The use of boto3 in celery#1759 resulted in relying on blocking (synchronous) HTTP requests, which caused the performance issue reported in celery#1783. `kombu` previously used to craft AWS requests manually as explained in detail in celery#1726, which resulted in an outage when botocore temporarily changed the default protocol to JSON (before rolling back due to the impact on celery and airflow.) To fix the issue, I submitted celery#1759, which changes `kombu` to use `boto3` instead of manually crafting AWS requests. This way when boto3 changes the default protocol, kombu won't be impacted. While working on celery#1759, I did extensive debugging to understand the multi-threading nature of kombu. What I discovered is that there isn't an actual multi-threading in the true sense of the word, but an event loop that runs on the same thread and process and orchestrate the communication with SQS. As such, it didn't appear to me that there is anything to worry about my change, and the testing I did didn't discover any issue. However, it turns out that while kombu's event loop doesn't have actual multi-threading, its [reliance on pycurl](https://github.com/celery/kombu/blob/main/kombu/asynchronous/http/curl.py#L48) (and thus libcurl) meant that the requests to AWS were being done asynchronously. On the other hand, boto3 requests are always done synchronously, i.e. they are blocking requests. The above meant that my change introduced blocking on the event loop of kombu. This is fine in most of the cases, since the requests to SQS are pretty fast. However, in the case of using long-polling, a call to SQS's ReceiveMessage can last up to 20 seconds (depending on the user configuration). To solve this problem, I rolled back my earlier changes and, instead, to address the issue reported in celery#1726, I now borrowed the implementation of `get_response` from botocore and changed the code such that the protocol is hard-coded to `query`. This way, when botocore changes the default protocol of SQS to JSON, kombu won't be impacted, since it crafts its own request and, after my change, it uses a hard-coded protocol based on the crafted requests. This solution shouldn't be the final solution, and it is more of a workaround that does the job for now. There are two problems with this approach: 1. It doesn't address the fundamental problem discussed in celery#1726, which is that `kombu` is using some stuff that are kind of internal to botocore, namely the `StreamingBody` class. 2. It is still making an assumption, namely that the protocol of communication is the `query` protocol. While this is true, and likely going to be true for some time, in theory nothing stops SQS (the backend, not client) from changing the default protocol, rendering the hard-coding of protocol in the new `get_response` method to `query` to be problematic. As such, the long term solution should be to completely rely on boto3 for any communication with AWS, and ensuring that all requests all async in nature (non-blocking.) This, however, is a fundamental change that requires a lot of testing, in particular performance testing.

This reverts commit 862d0bc.

TL;DR - The use of boto3 in celery#1759 resulted in relying on blocking (synchronous) HTTP requests, which caused the performance issue reported in celery#1783. `kombu` previously used to craft AWS requests manually as explained in detail in celery#1726, which resulted in an outage when botocore temporarily changed the default protocol to JSON (before rolling back due to the impact on celery and airflow.) To fix the issue, I submitted celery#1759, which changes `kombu` to use `boto3` instead of manually crafting AWS requests. This way when boto3 changes the default protocol, kombu won't be impacted. While working on celery#1759, I did extensive debugging to understand the multi-threading nature of kombu. What I discovered is that there isn't an actual multi-threading in the true sense of the word, but an event loop that runs on the same thread and process and orchestrate the communication with SQS. As such, it didn't appear to me that there is anything to worry about my change, and the testing I did didn't discover any issue. However, it turns out that while kombu's event loop doesn't have actual multi-threading, its [reliance on pycurl](https://github.com/celery/kombu/blob/main/kombu/asynchronous/http/curl.py#L48) (and thus libcurl) meant that the requests to AWS were being done asynchronously. On the other hand, boto3 requests are always done synchronously, i.e. they are blocking requests. The above meant that my change introduced blocking on the event loop of kombu. This is fine in most of the cases, since the requests to SQS are pretty fast. However, in the case of using long-polling, a call to SQS's ReceiveMessage can last up to 20 seconds (depending on the user configuration). To solve this problem, I rolled back my earlier changes and, instead, to address the issue reported in celery#1726, I now borrowed the implementation of `get_response` from botocore and changed the code such that the protocol is hard-coded to `query`. This way, when botocore changes the default protocol of SQS to JSON, kombu won't be impacted, since it crafts its own request and, after my change, it uses a hard-coded protocol based on the crafted requests. This solution shouldn't be the final solution, and it is more of a workaround that does the job for now. There are two problems with this approach: 1. It doesn't address the fundamental problem discussed in celery#1726, which is that `kombu` is using some stuff that are kind of internal to botocore, namely the `StreamingBody` class. 2. It is still making an assumption, namely that the protocol of communication is the `query` protocol. While this is true, and likely going to be true for some time, in theory nothing stops SQS (the backend, not client) from changing the default protocol, rendering the hard-coding of protocol in the new `get_response` method to `query` to be problematic. As such, the long term solution should be to completely rely on boto3 for any communication with AWS, and ensuring that all requests all async in nature (non-blocking.) This, however, is a fundamental change that requires a lot of testing, in particular performance testing.

TL;DR - The use of boto3 in celery#1759 resulted in relying on blocking (synchronous) HTTP requests, which caused the performance issue reported in celery#1783. `kombu` previously used to craft AWS requests manually as explained in detail in celery#1726, which resulted in an outage when botocore temporarily changed the default protocol to JSON (before rolling back due to the impact on celery and airflow.) To fix the issue, I submitted celery#1759, which changes `kombu` to use `boto3` instead of manually crafting AWS requests. This way when boto3 changes the default protocol, kombu won't be impacted. While working on celery#1759, I did extensive debugging to understand the multi-threading nature of kombu. What I discovered is that there isn't an actual multi-threading in the true sense of the word, but an event loop that runs on the same thread and process and orchestrate the communication with SQS. As such, it didn't appear to me that there is anything to worry about my change, and the testing I did didn't discover any issue. However, it turns out that while kombu's event loop doesn't have actual multi-threading, its [reliance on pycurl](https://github.com/celery/kombu/blob/main/kombu/asynchronous/http/curl.py#L48) (and thus libcurl) meant that the requests to AWS were being done asynchronously. On the other hand, boto3 requests are always done synchronously, i.e. they are blocking requests. The above meant that my change introduced blocking on the event loop of kombu. This is fine in most of the cases, since the requests to SQS are pretty fast. However, in the case of using long-polling, a call to SQS's ReceiveMessage can last up to 20 seconds (depending on the user configuration). To solve this problem, I rolled back my earlier changes and, instead, to address the issue reported in celery#1726, I now changed the `AsyncSQSConnection` class such that it crafts either a `query` or a `json` request depending on the protocol used by the SQS client. Thus, when botocore changes the default protocol of SQS to JSON, kombu won't be impacted, since it crafts its own request and, after my change, it uses a hard-coded protocol based on the crafted requests. This solution shouldn't be the final solution, and it is more of a workaround that does the job for now. The final solution should be to completely rely on boto3 for any communication with AWS, and ensuring that all requests are async in nature (non-blocking.) This, however, is a fundamental change that requires a lot of testing, in particular performance testing.

* Use the correct protocol for SQS requests TL;DR - The use of boto3 in #1759 resulted in relying on blocking (synchronous) HTTP requests, which caused the performance issue reported in #1783. `kombu` previously used to craft AWS requests manually as explained in detail in #1726, which resulted in an outage when botocore temporarily changed the default protocol to JSON (before rolling back due to the impact on celery and airflow.) To fix the issue, I submitted #1759, which changes `kombu` to use `boto3` instead of manually crafting AWS requests. This way when boto3 changes the default protocol, kombu won't be impacted. While working on #1759, I did extensive debugging to understand the multi-threading nature of kombu. What I discovered is that there isn't an actual multi-threading in the true sense of the word, but an event loop that runs on the same thread and process and orchestrate the communication with SQS. As such, it didn't appear to me that there is anything to worry about my change, and the testing I did didn't discover any issue. However, it turns out that while kombu's event loop doesn't have actual multi-threading, its [reliance on pycurl](https://github.com/celery/kombu/blob/main/kombu/asynchronous/http/curl.py#L48) (and thus libcurl) meant that the requests to AWS were being done asynchronously. On the other hand, boto3 requests are always done synchronously, i.e. they are blocking requests. The above meant that my change introduced blocking on the event loop of kombu. This is fine in most of the cases, since the requests to SQS are pretty fast. However, in the case of using long-polling, a call to SQS's ReceiveMessage can last up to 20 seconds (depending on the user configuration). To solve this problem, I rolled back my earlier changes and, instead, to address the issue reported in #1726, I now changed the `AsyncSQSConnection` class such that it crafts either a `query` or a `json` request depending on the protocol used by the SQS client. Thus, when botocore changes the default protocol of SQS to JSON, kombu won't be impacted, since it crafts its own request and, after my change, it uses a hard-coded protocol based on the crafted requests. This solution shouldn't be the final solution, and it is more of a workaround that does the job for now. The final solution should be to completely rely on boto3 for any communication with AWS, and ensuring that all requests are async in nature (non-blocking.) This, however, is a fundamental change that requires a lot of testing, in particular performance testing. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Update kombu/asynchronous/aws/sqs/connection.py * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Asif Saif Uddin <auvipy@gmail.com>

rafidka mentioned this pull request Jun 22, 2023

[Bug] AWS Connection.py Using Private Botocore get_response Which Expects JSON But is Being Passed XML #1726

Closed

auvipy self-requested a review June 22, 2023 02:57

auvipy requested changes Jun 22, 2023

View reviewed changes

rafidka force-pushed the fix-1726 branch from 5d93e2c to 8d5faf2 Compare June 23, 2023 00:02

rafidka requested a review from auvipy June 23, 2023 00:11

auvipy added the Component: Amazon SQS Broker label Jun 23, 2023

auvipy reviewed Jun 23, 2023

View reviewed changes

dhegberg approved these changes Jun 23, 2023

View reviewed changes

Kalhan22 approved these changes Jun 26, 2023

View reviewed changes

auvipy added this to the 5.3.x milestone Jun 26, 2023

rafidka force-pushed the fix-1726 branch 2 times, most recently from e117893 to f85d2c7 Compare June 26, 2023 13:13

[fix celery#1726] Use boto3 for SQS async requests

0e59389

rafidka force-pushed the fix-1726 branch from 08982cc to 0e59389 Compare June 26, 2023 13:28

[pre-commit.ci] auto fixes from pre-commit.com hooks

4c83711

for more information, see https://pre-commit.ci

rafidka requested a review from auvipy June 26, 2023 13:55

auvipy approved these changes Jun 27, 2023

View reviewed changes

auvipy added the PR Type: Cleanup label Jun 27, 2023

auvipy merged commit 862d0bc into celery:main Jun 27, 2023
14 checks passed

auvipy added a commit that referenced this pull request Sep 26, 2023

Revert "[fix #1726] Use boto3 for SQS async requests (#1759)"

cabe119

This reverts commit 862d0bc.

auvipy mentioned this pull request Sep 26, 2023

Revert "[fix #1726] Use boto3 for SQS async requests" #1799

Merged

rafidka mentioned this pull request Oct 4, 2023

Slow SQS task processing due to wait_time_in_seconds regression v5.3.2 #1783

Closed

rafidka added a commit to rafidka/kombu that referenced this pull request Oct 6, 2023

Revert "[fix celery#1726] Use boto3 for SQS async requests (celery#1759…

ff2c474

…)" This reverts commit 862d0bc.

rafidka mentioned this pull request Oct 9, 2023

Use the correct protocol for SQS requests #1807

Merged

auvipy added a commit that referenced this pull request Oct 10, 2023

Revert "[fix #1726] Use boto3 for SQS async requests (#1759)" (#1799)

1dfe4f3

This reverts commit 862d0bc.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix #1726] Use boto3 for SQS async requests #1759

[fix #1726] Use boto3 for SQS async requests #1759

rafidka commented Jun 22, 2023

auvipy left a comment

rafidka commented Jun 22, 2023

auvipy commented Jun 22, 2023

rafidka commented Jun 22, 2023

rafidka commented Jun 23, 2023

auvipy commented Jun 23, 2023

auvipy left a comment

rafidka commented Jun 23, 2023

rafidka commented Jun 26, 2023

auvipy commented Jun 27, 2023

auvipy commented Jun 27, 2023

auvipy commented Sep 18, 2023

auvipy commented Sep 26, 2023

rafidka commented Sep 26, 2023

auvipy commented Sep 27, 2023

rafidka commented Sep 27, 2023

auvipy commented Sep 27, 2023

rafidka commented Sep 29, 2023

rafidka commented Oct 4, 2023

[fix #1726] Use boto3 for SQS async requests #1759

[fix #1726] Use boto3 for SQS async requests #1759

Conversation

rafidka commented Jun 22, 2023

Overview

Testing

worker.py

scheduler.py

auvipy left a comment

Choose a reason for hiding this comment

rafidka commented Jun 22, 2023

auvipy commented Jun 22, 2023

rafidka commented Jun 22, 2023

rafidka commented Jun 23, 2023

auvipy commented Jun 23, 2023

auvipy left a comment

Choose a reason for hiding this comment

rafidka commented Jun 23, 2023

rafidka commented Jun 26, 2023

auvipy commented Jun 27, 2023

auvipy commented Jun 27, 2023

auvipy commented Sep 18, 2023

auvipy commented Sep 26, 2023

rafidka commented Sep 26, 2023

auvipy commented Sep 27, 2023

rafidka commented Sep 27, 2023

auvipy commented Sep 27, 2023

rafidka commented Sep 29, 2023

rafidka commented Oct 4, 2023