Issue receiving compressed requests

[oliviagyg]

Hi all,

I'm having issues getting compressed requests to process properly in my BentoML(0.13.1)-based service. Basically, I have a BentoService set up with a single endpoint that's configured to take StringInput (well, a subclass of StringInput). When I start up the dev server and send a request with a compressed body (and Content-Encoding: 'gzip', of course) it works perfectly. But when I try to do this with the gunicorn server, the service simply hangs, then times out.

I've poked around in the debugger, and it seems like the body gets decompressed before it ever makes it to 'from_http_request' in StringInput, which should actually be doing the decompression. I can go into why I think this is, but before I provide too much detail, I just want to ask: has anyone else noticed this behaviour? Is this a known issue?

[yubozhao]

Hey Olivia.

It is not a known issue. Would you mind to share more? We will also dig into it.

[oliviagyg]

Sure. This is probably more information than you really need, but here's a proof of concept with two files, one we'll call service.py:

from bentoml import BentoService, api
from bentoml.adapters import StringInput
class SampleService(BentoService):
    @api(input=StringInput(), route="sample_route", batch=False)
    def sample_endpoint(self, string_input: str):
        print(f"I successfully got the input: {string_input}")
        return {"returning the input": string_input}

if __name__ == '__main__':
    predictor = SampleService()
    predictor.save()

And another small file to send the requests, send_requests.py:

import requests
import gzip

def print_response(response):
    print(f"Status code: {response.status_code}")
    print(f"Content: {response.content}")

def send_requests():
    url = 'http://0.0.0.0:5000/sample_route'
    request_body = "just a plain old string"
    gzipped_request_body = gzip.compress(request_body.encode('utf-8'))
    print("First, a request without compression:")
    response = requests.post(url=url, data=request_body)
    print_response(response)
    print("\nNext, a request with compression:")
    header={'Content-Encoding': 'gzip'}
    response = requests.post(url=url, data=gzipped_request_body, headers=header)
    print_response(response)

if __name__ == '__main__':
    send_requests()

And I then run:

python service.py # Save the service bundle
python -m bentoml serve-gunicorn --debug SampleService:latest

And then, separately:

python send_requests.py

On the requesting side, I see:

First, a request without compression:
Status code: 200
Content: b'{"returning the input": "just a plain old string"}'

Next, a request with compression:
Status code: 503
Content: b'Service Unavailable'

And on the server, I get (after trimming down to the relevant portion):

[2022-05-05 16:37:34,402] DEBUG - Using BentoML default docker base image 'bentoml/model-server:0.13.1-py38'
[2022-05-05 16:37:40,353] DEBUG - Tracing is disabled. Initializing no-op tracer
[2022-05-05 16:37:40,381] DEBUG - Tracing is disabled. Initializing no-op tracer
I successfully got the input: just a plain old string
[2022-05-05 16:37:40,383] INFO - {'service_name': 'SampleService', 'service_version': '20220505163523_5C4A54', 'api': 'sample_endpoint', 'task': {'data': 'just a plain old string', 'task_id': '549e8c74-70ad-46be-8b78-217b3cea1521', 'http_headers': (('Host', '0.0.0.0:5000'), ('User-Agent', 'python-requests/2.27.1'), ('Accept-Encoding', 'gzip, deflate'), ('Accept', '*/*'), ('Connection', 'keep-alive'), ('Content-Length', '23'), ('Content-Type', 'application/octet-stream'))}, 'result': {'data': '{"returning the input": "just a plain old string"}', 'http_status': 200, 'http_headers': (('Content-Type', 'application/json'),)}, 'request_id': '549e8c74-70ad-46be-8b78-217b3cea1521'}
[2022-05-05 16:38:40 +0200] [21809] [CRITICAL] WORKER TIMEOUT (pid:21812)
[2022-05-05 16:38:40 +0200] [21812] [INFO] Worker exiting (pid: 21812)
[2022-05-05 16:38:40 +0200] [21849] [INFO] Booting worker with pid: 21849

Notably, if I send the request through curl instead (where ./tempzipped is a file with gzipped content):

curl --location --request POST 'http://0.0.0.0:5000/sample_route' \
--header 'Content-Encoding: gzip' \
--data-binary '@./tempzipped'

Then I actually get a stacktrace in the server output:

[2022-05-05 16:58:47 +0200] [21998] [CRITICAL] WORKER TIMEOUT (pid:22049)
[2022-05-05 16:58:47,910] ERROR - Traceback (most recent call last):
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/marshal/marshal.py", line 109, in request_dispatcher
    resp = await func(request)
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/marshal/marshal.py", line 338, in request_dispatcher
    resp = await self.relay_handler(request)
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/marshal/marshal.py", line 355, in relay_handler
    async with client.request(
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/aiohttp/client.py", line 1138, in __aenter__
    self._resp = await self._coro
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/aiohttp/client.py", line 559, in _request
    await resp.start(conn)
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/aiohttp/client_reqrep.py", line 913, in start
    self._continue = None
  File "/Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/aiohttp/helpers.py", line 721, in __exit__
    raise asyncio.TimeoutError from None
asyncio.exceptions.TimeoutError

So, if I stop the server and start it again in the debugger, and place a breakpoint in marshal.py, then run the curl command from above:

python -m pdb -m bentoml serve-gunicorn --debug SampleService:latest
> /Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/__main__.py(1)<module>()
-> if __name__ == '__main__':
(Pdb) b /Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/marshal/marshal.py:356
Breakpoint 1 at /Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/marshal/marshal.py:356
(Pdb) c
...
(skipping a bunch of output here)
...
> /Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/marshal/marshal.py(356)relay_handler()
-> request.method, url, data=data, headers=request.headers
(Pdb) print(data)
b'arbitrary string'
(Pdb) print(request.headers)
<CIMultiDictProxy('Host': '0.0.0.0:5000', 'User-Agent': 'curl/7.64.1', 'Accept': '*/*', 'Content-Encoding': 'gzip', 'Content-Length': '36', 'Content-Type': 'application/x-www-form-urlencoded')>

...I can see the headers and data at this point in the stack. Notably, the body has been decompressed and Content-Encoding is still 'gzip'.

I can also do the same thing, but running send_requests.py instead of the curl command to get a side-by-side of the non-compressed and compressed request:

(Pdb) print(data)
b'just a plain old string'
(Pdb) print(request.headers)
<CIMultiDictProxy('Host': '0.0.0.0:5000', 'User-Agent': 'python-requests/2.27.1', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '23')>
(Pdb) c
...
(Pdb) print(data)
b'just a plain old string'
(Pdb) print(request.headers)
<CIMultiDictProxy('Host': '0.0.0.0:5000', 'User-Agent': 'python-requests/2.27.1', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Encoding': 'gzip', 'Content-Length': '43')>

And finally, if I send a request that has a smaller compressed body than decompressed, then I don't actually see a timeout, but instead:

curl --location --request POST 'http://0.0.0.0:5000/sample_route' \
--header 'Content-Encoding: gzip' \
--data-binary '@./tempzipped2'

Gzip decompression error%    

..."Gzip decompression error", which matches bentoml -> adapters -> utils line 125, in decompress_gzip_request. And we can check what data and headers look like in the debugger:

(Pdb) print(data)
b'this is a pretty long string that will likely have a much smaller compressed size than uncompressed size'
(Pdb) print(request.headers)
<CIMultiDictProxy('Host': '0.0.0.0:5000', 'User-Agent': 'curl/7.64.1', 'Accept': '*/*', 'Content-Encoding': 'gzip', 'Content-Length': '93', 'Content-Type': 'application/x-www-form-urlencoded')>

I'm guessing that in the prior cases there's some issue with the content-length being shorter than the decompressed body, and that causes the service to hang while it's waiting for further input. And then for this case, with the content-length larger than the decompressed body, we actually make it to the StringInput portion, but decompression fails because the body has already been decompressed. That said, I'm just guessing based on what I see here, not super familiar with the architecture.

[timliubentoml]

Wow, thanks for the great detailed description. We will look into it and try to reproduce.

Is there any chance you could look into trying this same prediction endpoint out with our 1.0 version?

[oliviagyg]

Sure. I changed to bentoml 1.0.0a7 and updated service.py as follows:

import bentoml
from bentoml.io import Text
svc = bentoml.Service("SampleService")

@svc.api(input=Text(), output=Text(), route="sample_route")
def sample_endpoint(string_input: str):
    print(f"I successfully got the input: {string_input}")
    return f"returning the input: {string_input}"

Non-compressed requests work fine, but sending a compressed request shows:

[cli] Exception on /sample_route [POST] (trace=144464861372685485195197057743316861944,span=6255245580920166232,sampled=0)                       
╭────────────────────────────────────────────────────── Traceback (most recent call last) ──────────────────────────────────────────────────────╮
│ /Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/_internal/server/service_app.py:315 in api_func        │
│                                                                                                                                               │
│   312 │   │   ) -> "Response":                                                                                                                │
│   313 │   │   │   # handle_request may raise 4xx or 5xx exception.                                                                            │
│   314 │   │   │   try:                                                                                                                        │
│ ❱ 315 │   │   │   │   input_data = await api.input.from_http_request(request)                                                                 │
│   316 │   │   │   │   if asyncio.iscoroutinefunction(api.func):                                                                               │
│   317 │   │   │   │   │   if isinstance(api.input, Multipart):                                                                                │
│   318 │   │   │   │   │   │   output = await api.func(**input_data)                                                                           │
│                                                                                                                                               │
│ /Users/olivia.houghton/opt/miniconda3/envs/testenv/lib/python3.8/site-packages/bentoml/_internal/io_descriptors/text.py:88 in                 │
│ from_http_request                                                                                                                             │
│                                                                                                                                               │
│   85 │                                                                                                                                        │
│   86 │   async def from_http_request(self, request: Request) -> str:                                                                          │
│   87 │   │   obj = await request.body()                                                                                                       │
│ ❱ 88 │   │   return str(obj.decode("utf-8"))                                                                                                  │
│   89 │                                                                                                                                        │
│   90 │   async def to_http_response(self, obj: str) -> Response:                                                                              │
│   91 │   │   return Response(obj, media_type=MIME_TYPE)                                                                                       │
╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
UnicodeDecodeError: 'utf-8' codec can't decode byte 0x8b in position 1: invalid start byte

...in the server logs. Which is pretty clear: there just isn't handling for a gzipped request body in the Text IODescriptor. Which is fine -- I can just subclass Text and add that in, no problem:

import bentoml
import gzip
from bentoml.io import Text
svc = bentoml.Service("SampleService")

class MaybeCompressedText(Text):
    async def from_http_request(self, request) -> str:
        obj = await request.body()  
        headers = request.headers
        print(dir(headers))
        if headers.get('Content-Encoding') == 'gzip':
            obj = gzip.decompress(obj)
        return str(obj.decode('utf-8'))
        return obj

@svc.api(input=MaybeCompressedText(), output=Text(), route="sample_route")
def sample_endpoint(string_input: str):
    print(f"I successfully got the input: {string_input}")
    return f"returning the input: {string_input}"

And this does work for both compressed and non-compressed text, which is great!

[bojiang]

@oliviagyg Hi.
The API server of bentoml 1.0 doesn't support compressed requests by default currently, but users can customize it easily with the service.add_moddleware API. And it is also the recommended approach.
For example, to support gzip-compressed HTTP requests, we can just add the GZipMiddleware of Starlette (https://www.starlette.io/middleware/#GZipMiddleware) by simply calling svc.add_middleware(GZipMiddleware ) at the end of the service.py