asyncio support

[aviklai]

13

Couple of open issues:

1. Need to understand how to compute the status code as _state is undefined for the servicer_context.
2. The tests will need a different setup as we will need an aio grpc server which we can't start with pytest.

[tesserata]

Hello, @aviklai !

I've been trying to use the async version you've provided in this fork and have been having some difficulties in making it work, would you be able to provide any ideas of what may be the cause?

I use a standard async grpc with reflection:

import asyncio
import grpc
from grpc_reflection.v1alpha import reflection
from concurrent import futures
from py_grpc_prometheus.prometheus_aio_server_interceptor import PromAioServerInterceptor

server = grpc.aio.server(
    futures.ThreadPoolExecutor(max_workers=1000),
    options=[
        ('grpc.max_send_message_length', 400000000),
        ('grpc.max_receive_message_length', 400000000)
    ],
    interceptors=(PromAioServerInterceptor(legacy=True),)
)
start_http_server(9090)
api_pb2_grpc.add_DWHApiServicer_to_server(dwhrc_servicer, server)
service_names = (
    api_pb2.DESCRIPTOR.services_by_name['DWHApi'].full_name,
    reflection.SERVICE_NAME,
)
reflection.enable_server_reflection(service_names, server)
server.add_insecure_port('[::]:50051')

However, when trying to execute a request (sent via grpcurl), the server fails with the following error:

2023-03-26 12:31:30.887 | ERROR    | asyncio.events:_run:81 - Unexpected [TypeError] raised by servicer method [/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo]
Traceback (most recent call last):

  File "/app/server.py", line 139, in <module>
    asyncio.run(serve())
    │       │   └ <function serve at 0x7fc1981ff1f0>
    │       └ <function run at 0x7fc1a6c028b0>
    └ <module 'asyncio' from '/usr/local/lib/python3.8/asyncio/__init__.py'>

  File "/usr/local/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
           │    │                  └ <coroutine object serve at 0x7fc198491c40>
           │    └ <function BaseEventLoop.run_until_complete at 0x7fc1a6482b80>
           └ <_UnixSelectorEventLoop running=True closed=False debug=False>
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 603, in run_until_complete
    self.run_forever()
    │    └ <function BaseEventLoop.run_forever at 0x7fc1a6482af0>
    └ <_UnixSelectorEventLoop running=True closed=False debug=False>
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 570, in run_forever
    self._run_once()
    │    └ <function BaseEventLoop._run_once at 0x7fc1a6484670>
    └ <_UnixSelectorEventLoop running=True closed=False debug=False>
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 1859, in _run_once
    handle._run()
    │      └ <function Handle._run at 0x7fc1a6515430>
    └ <Handle <TaskStepMethWrapper object at 0x7fc1a4e9b220>()>
  File "/usr/local/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
    │    │            │    │           │    └ <member '_args' of 'Handle' objects>
    │    │            │    │           └ <Handle <TaskStepMethWrapper object at 0x7fc1a4e9b220>()>
    │    │            │    └ <member '_callback' of 'Handle' objects>
    │    │            └ <Handle <TaskStepMethWrapper object at 0x7fc1a4e9b220>()>
    │    └ <member '_context' of 'Handle' objects>
    └ <Handle <TaskStepMethWrapper object at 0x7fc1a4e9b220>()>
> File "src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi", line 682, in grpc._cython.cygrpc._handle_exceptions
  File "src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi", line 817, in _handle_rpc
  File "src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi", line 670, in _handle_stream_stream_rpc
  File "src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi", line 472, in _finish_handler_with_stream_responses
  File "/usr/local/lib/python3.8/site-packages/py_grpc_prometheus/prometheus_aio_server_interceptor.py", line 113, in new_behavior
    raise e
  File "/usr/local/lib/python3.8/site-packages/py_grpc_prometheus/prometheus_aio_server_interceptor.py", line 63, in new_behavior
    response_or_iterator = await behavior(request_or_iterator, servicer_context)
                                 │        │                    └ <grpc._cython.cygrpc._ServicerContext object at 0x7fc174e4c940>
                                 │        └ <generator object wrap_iterator_inc_counter at 0x7fc17494d510>
                                 └ <bound method ReflectionServicer.ServerReflectionInfo of <grpc_reflection.v1alpha._async.ReflectionServicer object at 0x7fc19...

TypeError: object async_generator can't be used in 'await' expression

[aviklai]

@tesserata Hi,

Never tried this with server reflection. Can you try it without server reflection?

[Roktober]

@aviklai I wrote an implementation based on yours. After studying the implementation of metrics in other languages, I found differences in the calculation of metrics, please pay attention to my implementation and the difference with yours.

Now at the stage of writing tests.

https://gist.github.com/Roktober/61140e1d817e565d8521082d9a9bba22

[aviklai]

@Roktober Hi,

Thanks but the metrics calculation is not my implementation. I just applied the same implementation in this repo with the async capability support. Maybe you can create a pull request to this repo with the relevant changes?