feat: enable HTTP and Composite protocols for Deployment (jina-ai#5764)

Signed-off-by: Joan Fontanals Martinez <[email protected]>
Co-authored-by: Jina Dev Bot <[email protected]>

Author: JoanFM

.github/workflows/cd.yml

@@ -359,7 +359,7 @@ jobs:
         run: |
           export LINKERD2_VERSION=stable-2.11.4
           curl --proto '=https' --tlsv1.2 -sSfL https://run.linkerd.io/install | sh
-          pytest -v -s --suppress-no-test-exit-code --force-flaky --min-passes 1 --max-runs 5 --cov=jina --cov-report=xml ./tests/k8s/test_k8s_deployment.py ./tests/k8s/test_graceful_request_handling.py
+          pytest -v -s --suppress-no-test-exit-code --force-flaky --min-passes 1 --max-runs 5 --cov=jina --cov-report=xml ./tests/k8s/test_k8s_deployment.py ./tests/k8s/test_k8s_graceful_request_handling.py
         timeout-minutes: 30
       - name: Check codecov file
         id: check_files

.github/workflows/ci.yml

@@ -232,7 +232,7 @@ jobs:
         run: |
           export LINKERD2_VERSION=stable-2.11.4
           curl --proto '=https' --tlsv1.2 -sSfL https://run.linkerd.io/install | sh
-          pytest -v -s --suppress-no-test-exit-code --force-flaky --min-passes 1 --max-runs 5 --cov=jina --cov-report=xml ./tests/k8s/test_k8s_deployment.py ./tests/k8s/test_graceful_request_handling.py
+          pytest -v -s --suppress-no-test-exit-code --force-flaky --min-passes 1 --max-runs 5 --cov=jina --cov-report=xml ./tests/k8s/test_k8s_deployment.py ./tests/k8s/test_k8s_graceful_request_handling.py
         timeout-minutes: 45
       - name: Check codecov file
         id: check_files

docs/concepts/flow/deployment-args.md

@@ -36,6 +36,7 @@
 | `reload` | If set, the Executor will restart while serving if YAML configuration source or Executor modules are changed. If YAML configuration is changed, the whole deployment is reloaded and new processes will be restarted. If only Python modules of the Executor have changed, they will be reloaded to the interpreter without restarting process. | `boolean` | `False` |
 | `install_requirements` | If set, try to install `requirements.txt` from the local Executor if exists in the Executor folder. If using Hub, install `requirements.txt` in the Hub Executor bundle to local. | `boolean` | `False` |
 | `port` | The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports. Then, every resulting address will be considered as one replica of the Executor. | `number` | `random in [49152, 65535]` |
+| `protocol` | Communication protocol of the server exposed by the Executor. This can be a single value or a list of protocols, depending on your chosen Gateway. Choose the convenient protocols from: ['GRPC', 'HTTP', 'WEBSOCKET']. | `array` | `[<ProtocolType.GRPC: 0>]` |
 | `monitoring` | If set, spawn an http server with a prometheus endpoint to expose metrics | `boolean` | `False` |
 | `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
 | `retries` | Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas) | `number` | `-1` |
@@ -57,4 +58,10 @@
 | `when` | The condition that the documents need to fulfill before reaching the Executor.The condition can be defined in the form of a `DocArray query condition <https://docarray.jina.ai/fundamentals/documentarray/find/#query-by-conditions>` | `object` | `None` |
 | `external` | The Deployment will be considered an external Deployment that has been started independently from the Flow.This Deployment will not be context managed by the Flow. | `boolean` | `False` |
 | `grpc_metadata` | The metadata to be passed to the gRPC request. | `object` | `None` |
-| `tls` | If set, connect to deployment using tls encryption | `boolean` | `False` |
+| `tls` | If set, connect to deployment using tls encryption | `boolean` | `False` |
+| `title` | The title of this HTTP server. It will be used in automatics docs such as Swagger UI. | `string` | `None` |
+| `description` | The description of this HTTP server. It will be used in automatics docs such as Swagger UI. | `string` | `None` |
+| `cors` | If set, a CORS middleware is added to FastAPI frontend to allow cross-origin access. | `boolean` | `False` |
+| `uvicorn_kwargs` | Dictionary of kwargs arguments that will be passed to Uvicorn server when starting the server<br><br>More details can be found in Uvicorn docs: https://www.uvicorn.org/settings/ | `object` | `None` |
+| `ssl_certfile` | the path to the certificate file | `string` | `None` |
+| `ssl_keyfile` | the path to the key file | `string` | `None` |

docs/concepts/flow/executor-args.md

@@ -36,6 +36,7 @@
 | `reload` | If set, the Executor will restart while serving if YAML configuration source or Executor modules are changed. If YAML configuration is changed, the whole deployment is reloaded and new processes will be restarted. If only Python modules of the Executor have changed, they will be reloaded to the interpreter without restarting process. | `boolean` | `False` |
 | `install_requirements` | If set, try to install `requirements.txt` from the local Executor if exists in the Executor folder. If using Hub, install `requirements.txt` in the Hub Executor bundle to local. | `boolean` | `False` |
 | `port` | The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports. Then, every resulting address will be considered as one replica of the Executor. | `number` | `random in [49152, 65535]` |
+| `protocol` | Communication protocol of the server exposed by the Executor. This can be a single value or a list of protocols, depending on your chosen Gateway. Choose the convenient protocols from: ['GRPC', 'HTTP', 'WEBSOCKET']. | `array` | `[<ProtocolType.GRPC: 0>]` |
 | `monitoring` | If set, spawn an http server with a prometheus endpoint to expose metrics | `boolean` | `False` |
 | `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
 | `retries` | Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas) | `number` | `-1` |

docs/concepts/flow/gateway-args.md

@@ -12,14 +12,13 @@
 | `title` | The title of this HTTP server. It will be used in automatics docs such as Swagger UI. | `string` | `None` |
 | `description` | The description of this HTTP server. It will be used in automatics docs such as Swagger UI. | `string` | `None` |
 | `cors` | If set, a CORS middleware is added to FastAPI frontend to allow cross-origin access. | `boolean` | `False` |
-| `no_debug_endpoints` | If set, `/status` `/post` endpoints are removed from HTTP interface. | `boolean` | `False` |
-| `no_crud_endpoints` | If set, `/index`, `/search`, `/update`, `/delete` endpoints are removed from HTTP interface.<br><br>        Any executor that has `@requests(on=...)` bound with those values will receive data requests. | `boolean` | `False` |
-| `expose_endpoints` | A JSON string that represents a map from executor endpoints (`@requests(on=...)`) to HTTP endpoints. | `string` | `None` |
 | `uvicorn_kwargs` | Dictionary of kwargs arguments that will be passed to Uvicorn server when starting the server<br><br>More details can be found in Uvicorn docs: https://www.uvicorn.org/settings/ | `object` | `None` |
 | `ssl_certfile` | the path to the certificate file | `string` | `None` |
 | `ssl_keyfile` | the path to the key file | `string` | `None` |
+| `no_debug_endpoints` | If set, `/status` `/post` endpoints are removed from HTTP interface. | `boolean` | `False` |
+| `no_crud_endpoints` | If set, `/index`, `/search`, `/update`, `/delete` endpoints are removed from HTTP interface.<br><br>        Any executor that has `@requests(on=...)` bound with those values will receive data requests. | `boolean` | `False` |
+| `expose_endpoints` | A JSON string that represents a map from executor endpoints (`@requests(on=...)`) to HTTP endpoints. | `string` | `None` |
 | `expose_graphql_endpoint` | If set, /graphql endpoint is added to HTTP interface. | `boolean` | `False` |
-| `protocol` | Communication protocol of the server exposed by the Gateway. This can be a single value or a list of protocols, depending on your chosen Gateway. Choose the convenient protocols from: ['GRPC', 'HTTP', 'WEBSOCKET']. | `array` | `[<GatewayProtocolType.GRPC: 0>]` |
 | `host` | The host address of the runtime, by default it is 0.0.0.0. | `string` | `0.0.0.0` |
 | `proxy` | If set, respect the http_proxy and https_proxy environment variables. otherwise, it will unset these proxy variables before start. gRPC seems to prefer no proxy | `boolean` | `False` |
 | `uses` | The config of the gateway, it could be one of the followings:<br>        * the string literal of an Gateway class name<br>        * a Gateway YAML file (.yml, .yaml, .jaml)<br>        * a docker image (must start with `docker://`)<br>        * the string literal of a YAML config (must start with `!` or `jtype: `)<br>        * the string literal of a JSON config<br><br>        When use it under Python, one can use the following values additionally:<br>        - a Python dict that represents the config<br>        - a text file stream has `.read()` interface | `string` | `None` |
@@ -42,6 +41,7 @@
 | `floating` | If set, the current Pod/Deployment can not be further chained, and the next `.add()` will chain after the last Pod/Deployment not this current one. | `boolean` | `False` |
 | `reload` | If set, the Gateway will restart while serving if YAML configuration source is changed. | `boolean` | `False` |
 | `port` | The port for input data to bind the gateway server to, by default, random ports between range [49152, 65535] will be assigned. The port argument can be either 1 single value in case only 1 protocol is used or multiple values when many protocols are used. | `number` | `random in [49152, 65535]` |
+| `protocol` | Communication protocol of the server exposed by the Gateway. This can be a single value or a list of protocols, depending on your chosen Gateway. Choose the convenient protocols from: ['GRPC', 'HTTP', 'WEBSOCKET']. | `array` | `[<ProtocolType.GRPC: 0>]` |
 | `monitoring` | If set, spawn an http server with a prometheus endpoint to expose metrics | `boolean` | `False` |
 | `port_monitoring` | The port on which the prometheus server is exposed, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
 | `retries` | Number of retries per gRPC call. If <0 it defaults to max(3, num_replicas) | `number` | `-1` |

jina/__init__.py

@@ -65,7 +65,7 @@ def _warning_on_one_line(message, category, filename, lineno, *args, **kwargs):
 
 # do not change this line manually
 # this is managed by proto/build-proto.sh and updated on every execution
-__proto_version__ = '0.1.13'
+__proto_version__ = '0.1.17'
 
 try:
     __docarray_version__ = _docarray.__version__

jina/checker.py

@@ -1,6 +1,6 @@
 import argparse
 
-from jina.enums import GatewayProtocolType
+from jina.enums import ProtocolType
 from jina.helper import parse_host_scheme
 from jina.logging.predefined import default_logger
 

jina/clients/__init__.py

@@ -6,7 +6,7 @@
 
 __all__ = ['Client']
 
-from jina.enums import GatewayProtocolType
+from jina.enums import ProtocolType
 
 if TYPE_CHECKING:  # pragma: no cover
     from jina.clients.grpc import AsyncGRPCClient, GRPCClient
@@ -132,15 +132,13 @@ def Client(
     ):  # we need to parse the kwargs as soon as possible otherwise to get the gateway type
         args = parse_client(kwargs)
 
-    protocol = (
-        args.protocol if args else kwargs.get('protocol', GatewayProtocolType.GRPC)
-    )
+    protocol = args.protocol if args else kwargs.get('protocol', ProtocolType.GRPC)
     if isinstance(protocol, str):
-        protocol = GatewayProtocolType.from_string(protocol)
+        protocol = ProtocolType.from_string(protocol)
 
     is_async = (args and args.asyncio) or kwargs.get('asyncio', False)
 
-    if protocol == GatewayProtocolType.GRPC:
+    if protocol == ProtocolType.GRPC:
         if is_async:
             from jina.clients.grpc import AsyncGRPCClient
 
@@ -149,7 +147,7 @@ def Client(
             from jina.clients.grpc import GRPCClient
 
             return GRPCClient(args, **kwargs)
-    elif protocol == GatewayProtocolType.WEBSOCKET:
+    elif protocol == ProtocolType.WEBSOCKET:
         if is_async:
             from jina.clients.websocket import AsyncWebSocketClient
 
@@ -158,7 +156,7 @@ def Client(
             from jina.clients.websocket import WebSocketClient
 
             return WebSocketClient(args, **kwargs)
-    elif protocol == GatewayProtocolType.HTTP:
+    elif protocol == ProtocolType.HTTP:
         if is_async:
             from jina.clients.http import AsyncHTTPClient
 

jina/clients/base/http.py

@@ -18,6 +18,34 @@
 class HTTPBaseClient(BaseClient):
     """A MixIn for HTTP Client."""
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._endpoints = []
+
+    async def _get_endpoints_from_openapi(self):
+        def extract_paths_by_method(spec):
+            paths_by_method = {}
+            for path, methods in spec['paths'].items():
+                for method, details in methods.items():
+                    if method not in paths_by_method:
+                        paths_by_method[method] = []
+                    paths_by_method[method].append(path.strip('/'))
+
+            return paths_by_method
+        import aiohttp
+        import json
+
+        proto = 'https' if self.args.tls else 'http'
+        target_url = f'{proto}://{self.args.host}:{self.args.port}/openapi.json'
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(target_url) as response:
+                    content = await response.read()
+                    openapi_response = json.loads(content.decode())
+                    self._endpoints = extract_paths_by_method(openapi_response).get('post', [])
+        except:
+            pass
+
     async def _is_flow_ready(self, **kwargs) -> bool:
         """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
 
@@ -57,18 +85,18 @@ async def _is_flow_ready(self, **kwargs) -> bool:
         return False
 
     async def _get_results(
-        self,
-        inputs: 'InputType',
-        on_done: 'CallbackFnType',
-        on_error: Optional['CallbackFnType'] = None,
-        on_always: Optional['CallbackFnType'] = None,
-        max_attempts: int = 1,
-        initial_backoff: float = 0.5,
-        max_backoff: float = 0.1,
-        backoff_multiplier: float = 1.5,
-        results_in_order: bool = False,
-        prefetch: Optional[int] = None,
-        **kwargs,
+            self,
+            inputs: 'InputType',
+            on_done: 'CallbackFnType',
+            on_error: Optional['CallbackFnType'] = None,
+            on_always: Optional['CallbackFnType'] = None,
+            max_attempts: int = 1,
+            initial_backoff: float = 0.5,
+            max_backoff: float = 0.1,
+            backoff_multiplier: float = 1.5,
+            results_in_order: bool = False,
+            prefetch: Optional[int] = None,
+            **kwargs,
     ):
         """
         :param inputs: the callable
@@ -89,15 +117,26 @@ async def _get_results(
 
         self.inputs = inputs
         request_iterator = self._get_requests(**kwargs)
+        on = kwargs.get('on', '/post')
+        if len(self._endpoints) == 0:
+            await self._get_endpoints_from_openapi()
 
         async with AsyncExitStack() as stack:
             cm1 = ProgressBar(
-                total_length=self._inputs_length, disable=not (self.show_progress)
+                total_length=self._inputs_length, disable=not self.show_progress
             )
             p_bar = stack.enter_context(cm1)
-
             proto = 'https' if self.args.tls else 'http'
-            url = f'{proto}://{self.args.host}:{self.args.port}/post'
+            endpoint = on.strip('/')
+            has_default_endpoint = 'default' in self._endpoints
+
+            if endpoint != '' and endpoint in self._endpoints:
+                url = f'{proto}://{self.args.host}:{self.args.port}/{on.strip("/")}'
+            elif has_default_endpoint:
+                url = f'{proto}://{self.args.host}:{self.args.port}/default'
+            else:
+                url = f'{proto}://{self.args.host}:{self.args.port}/post'
+
             iolet = await stack.enter_async_context(
                 HTTPClientlet(
                     url=url,
@@ -112,7 +151,7 @@ async def _get_results(
             )
 
             def _request_handler(
-                request: 'Request',
+                    request: 'Request',
             ) -> 'Tuple[asyncio.Future, Optional[asyncio.Future]]':
                 """
                 For HTTP Client, for each request in the iterator, we `send_message` using
@@ -135,7 +174,7 @@ def _result_handler(result):
                 **streamer_args,
             )
             async for response in streamer.stream(
-                request_iterator=request_iterator, results_in_order=results_in_order
+                    request_iterator=request_iterator, results_in_order=results_in_order
             ):
                 r_status = response.status