Repository: jina-ai/serve
Pull request: feat: expose flow readiness check in gateway
This PR aims to expose a `health` endpoint to the gateway that works as a dry_run to expose to the client a way to know if the Flow is ready to receive requests. 

It is important to differentiate this `health` endpoint with the `healthiness check` of an individual runtime.

Maybe the name `health` or `health_check` should be changed in this context to some `dry_run`?


Problem statement:
add jina health check to jina ping
We want user to be able to use the jina health checks as now it is only a internal tools.

- [ ] document the jina health checks
- [ ] merge jina health check and jina ping (see further)

How to do health check from client?
`client.check?/.ping?`

Healthcheck endpoints for server.
Are there standard healthcheck endpoints as part of the server for both http and grpc? If not it would make it much easier to register a healthcheck for a docker container if this was added. Thank you for your hard work!

Hints from issue discussion:
@samsja What is this issue? I assume I cannot join slack, but I would be interested in familiarizing myself with these issues as I'm considering contributing here.
Hey, we are glad that you want to contribute ! You can definitely work on this issue if you want to

To give you some context:

We recently added some health check to jina ( see here) : https://github.com/jina-ai/jina/blob/84089a0287933ec07b165d32498abc92d6f95898/jina/resources/health_check/gateway.py#L33)
For example if you have setup a flow you can check if it is healthy by doing the following command:

```python
from jina import Flow, Client

if __name__ == '__main__':
    f = Flow(port=12345).add()
    with f:
        f.block()

```

```bash
python -m jina.resources.health_check.gateway localhost:12345 grpc
```

At the same time we have a `ping` option in the cli which kind of doing the same things so we would like the fuse both functionality so that 

```bash
jina ping localhost 12345 
```
call the health check under the hood:

it means that the `NetworkChecker` here https://github.com/jina-ai/jina/blob/84089a0287933ec07b165d32498abc92d6f95898/jina/checker.py#L6 need to be rewrite so that the ping command https://github.com/jina-ai/jina/blob/84089a0287933ec07b165d32498abc92d6f95898/cli/api.py#L120 is up to date with the health checking


Let me know if you want to work on this issue and if you need any help

There is not such a functionality, and I think this is not the job of fhe Client, this would mean mixing the orchestration part (which is Kubernetes job) and the data processing part which is indeed Jina's job.
in CaS, client provide a `profile` interface https://clip-as-service.jina.ai/user-guides/client/#profiling that allows you to "healthcheck" the server.

I guess your question is more about how to directly & natively check server status from K8s/cloud-native ecosystem. Jina core should have provided the interface, maybe it needs to be documented. Will move this issue to the core for better discussion.
Hey @MrBeeMovie,

How are you deploying Clip as a Service? Is it locally, on Kubernetes or with docker compose?


> Hey @MrBeeMovie,
> 
> How are you deploying Clip as a Service? Is it locally, on Kubernetes or with docker compose?

I am running on local. I want a direct way to check the server status because I am using the HEALTHCHECK keyword in the dockerfile. It would be useful to be able to curl localhost:port/healthcheck or something similar so I don't have to setup the clip-client and then call profile.
> in CaS, client provide a `profile` interface https://clip-as-service.jina.ai/user-guides/client/#profiling that allows you to "healthcheck" the server.
> 
> I guess your question is more about how to directly & natively check server status from K8s/cloud-native ecosystem. Jina core should have provided the interface, maybe it needs to be documented. Will move this issue to the core for better discussion.

Yeah, sorry if I missed this in the documentation. I will look over it again. I am looking for a way to curl an endpoint from clip-server so that I do not have to setup clip-client to check the server status. If you can direct me towards the appropriate documentation that would be great. I understand that this is probably more related to Jina core but maybe you can include in the documentation under [plain http requests](https://clip-as-service.jina.ai/user-guides/client/#profiling) if this endpoint does exist. Thank you for your time!
@MrBeeMovie An health check endpoint for `protocol: http` exists already at `GET /`.

![image](https://user-images.githubusercontent.com/9050737/166641584-1a51e55a-4299-44d2-9f4d-5032d26f0564.png)

But you're right, we need to add it for `protocol: grpc`. An example usecase is, AWS ALB expects a path to be defined for health check (defaults to `/AWS.ALB/healthcheck`). It also supports a [response code](https://grpc.github.io/grpc/core/md_doc_statuscodes.html), which defaults to `12 (UNIMPLEMENTED)`. This path should be provided by `jina` & the response code should be `0 (OK)`. 

![image](https://user-images.githubusercontent.com/9050737/166642225-a36a1e17-070b-4946-8427-72975f93c811.png)

The gRPC health check protocol is documented [here](https://github.com/grpc/grpc/blob/master/doc/health-checking.md#grpc-health-checking-protocol). 
The scope of this issue is to implement a grpc health check for every Jina pod as [defined by their specification](https://github.com/grpc/grpc/blob/master/doc/health-checking.md#grpc-health-checking-protocol).
This new endpoint should eventually replace the old ControlRequest STATUS endpoint. For compatibility reasons, I would keep the old endpoint around for a little bit though.

- [ ] Implement the new grpc health check
- [ ] Switch over Jina implementation to use the new endpoint, instead of the STATUS endpoint
- [ ] Document how the new health check can be used (and also mention the existing http version)
- [ ] Create and schedule an issue for the removal of the STATUS endpoint
Also @jacobowitz from this ticket, should we open a ticket about creating a `ready_check` that can be called from client?
no, I think that is unrelated and should be done [here](https://github.com/jina-ai/jina/issues/4730)

Commit message: docs: add health check documentation (#4848)

Patch to review:
diff --git a/docs/fundamentals/flow/index.md b/docs/fundamentals/flow/index.md
index 9a5c55257af8f..0609cb5accca1 100644
--- a/docs/fundamentals/flow/index.md
+++ b/docs/fundamentals/flow/index.md
@@ -22,6 +22,7 @@ The most important methods of the `Flow` object are the following:
 | `.block()`                         | Blocks execution until the program is terminated. This is useful to keep the Flow alive so it can be used from other places (clients, etc). |
 | `.to_docker_compose_yaml()`        | Generates a Docker-Compose file listing all its Executors as Services.                                                                       |
 | `.to_k8s_yaml(<output_directory>)` | Generates the Kubernetes configuration files in `<output_directory>`. Based on your local Jina version, Jina Hub may rebuild the Docker image during the YAML generation process. If you do not wish to rebuild the image, set the environment variable `JINA_HUB_NO_IMAGE_REBUILD`. |
+| `.dry_run()`                       | Calls the dry run endpoint of the Flow to check if the Flow is ready to process requests. Returns a boolean indicating the readiness |
 
 ## Why should you use a Flow?
 
diff --git a/docs/fundamentals/health-check.md b/docs/fundamentals/health-check.md
new file mode 100644
index 0000000000000..b6a4dddf29aba
--- /dev/null
+++ b/docs/fundamentals/health-check.md
@@ -0,0 +1,238 @@
+(health-check-executor)=
+# Health check of an Executor
+
+Executors run as microservices exposing `grpc` endpoints. To give information to the outside world about their health and their readiness to receive requests,
+Executors expose a [grpc health check](https://github.com/grpc/grpc/blob/master/doc/health-checking.md) service which can be used by different orchestrators like docker-compose, 
+or by other services like load-balancers. Jina itself when run locally uses this service to make sure that each Executor is ready to receive traffic.
+
+To see how to use it, let's start a Flow inside a Terminal and let it blocked to serve requests:
+
+```python
+from jina import Flow
+
+f = Flow(protocol='grpc', port=12345).add(port=12346)
+with f:
+    f.block()
+```
+
+On another terminal, we can install or download [grpcurl](https://github.com/fullstorydev/grpcurl) image to be able to send `rpc` requests to our services.
+
+```bash
+docker pull fullstorydev/grpcurl:latest
+docker run --network='host' fullstorydev/grpcurl -plaintext 127.0.0.1:12346 grpc.health.v1.Health/Check
+```
+
+```text
+{
+  "status": "SERVING"
+}
+```
+
+(health-check-gateway)=
+# Health check of the Gateway
+
+The same way individual Executors expose endpoints for orchestrators, clients or other services to check their availability, Gateway as a microservice also exposes this in different ways and depending
+on the protocol used.
+
+
+## Gateway health check with grpc
+
+When using grpc as the protocol used to communicate with the Gateway, then Gateway uses the exact same mechanism as Executors to expose their individual health status. It exposes [grpc health check](https://github.com/grpc/grpc/blob/master/doc/health-checking.md) to the outside world.
+
+With the same Flow described before we can use the same way to check the gateway status:
+
+```bash
+docker run --network='host' fullstorydev/grpcurl -plaintext 127.0.0.1:12345 grpc.health.v1.Health/Check
+```
+
+```text
+{
+  "status": "SERVING"
+}
+```
+
+
+## Gateway health check with http or websocket
+
+When using grpc or http as a procotol for the Gateway, then it exposes and endpoint '/' that one can query to check the status.
+
+By having a Flow with http or websocket as protocol:
+
+```python
+from jina import Flow
+
+f = Flow(protocol='http', port=12345).add()
+with f:
+    f.block()
+```
+
+```bash
+curl http://localhost:12345
+```
+
+Then you will get a valid empty response indicating its ability to serve.
+```text
+{}%
+```
+
+# Readiness of a Flow exposed to the client
+
+A lot of times, from the client perspective, it is useful to know if a Flow as a complete set of microservices is ready to receive requests. This is why the gateway 
+exposes an endpoint for each of the supported protocols to know the health and readiness of a Flow. 
+
+Also Jina Flow and Client contains a convenient API to query these endpoints. You can call `flow.dry_run()` and `client.dry_run()`
+
+````{tab} via Flow
+```python
+from jina import Flow
+
+with Flow().add() as f:
+    print(f.dry_run())
+
+print(f.dry_run())
+```
+```text
+True
+False
+```
+````
+````{tab} via Client
+```python
+from jina import Flow
+
+with Flow(port=12345).add() as f:
+    f.block()
+```
+```python
+from jina import Client
+
+client = Client(port=12345)
+print(client.dry_run())
+```
+```text
+True
+```
+````
+
+## Check status from outside
+
+To understand better what is going on under the hood, we can check how to target the endpoint from outside Jina or the client.
+
+Let's first instantiate the Flow with its corresponding protocol and block it for serving.
+
+```python
+from jina import Flow
+import os
+
+PROTOCOL = 'grpc'  # it could also be http or websocket
+
+os.setenv[
+    'JINA_LOG_LEVEL'
+] = 'DEBUG'  # this way we can check what is the PID of the Executor
+
+with Flow(protocol=PROTOCOL, port=12345).add() as f:
+    f.block()
+```
+
+```text
+⠋  Waiting ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0/0 -:--:--DEBUG  gateway/rep-0@19075 adding connection for deployment executor0/heads/0 to grpc://0.0.0.0:12346                                                                                           [05/31/22 18:10:16]
+DEBUG  executor0/rep-0@19074 start listening on 0.0.0.0:12346                                                                                                                                   [05/31/22 18:10:16]
+DEBUG  gateway/rep-0@19075 start server bound to 0.0.0.0:12345                                                                                                                                  [05/31/22 18:10:17]
+DEBUG  executor0/rep-0@19059 ready and listening                                                                                                                                                [05/31/22 18:10:17]
+DEBUG  gateway/rep-0@19059 ready and listening                                                                                                                                                  [05/31/22 18:10:17]
+╭────── 🎉 Flow is ready to serve! ──────╮
+│  🔗  Protocol                  GRPC    │
+│  🏠     Local         0.0.0.0:12345    │
+│  🔒   Private    192.168.1.13:12345    │
+│  🌍    Public  92.178.210.177:12345    │
+╰────────────────────────────────────────╯
+DEBUG  Flow@19059 2 Deployments (i.e. 2 Pods) are running in this Flow 
+```
+
+### with grpc
+
+When using grpc, we can again use grpcurl to hit the proper grpc service, we will see that the Status is positive.
+
+```shell
+docker run --network='host' fullstorydev/grpcurl -plaintext 127.0.0.1:12345 jina.JinaGatewayDryRunRPC/dry_run
+```
+```text
+{
+  
+}
+```
+
+Let's simulate an Executor going offline by killing its process.
+
+```shell script
+kill -9 $EXECUTOR_PID # in this case we can see in the logs that it is 19059
+```
+
+Then by doing the same check, we will see that it returns an error.
+
+```shell
+docker run --network='host' fullstorydev/grpcurl -plaintext 127.0.0.1:12345 jina.JinaGatewayDryRunRPC/dry_run
+```
+
+```text
+{
+  "code": "ERROR",
+  "description": "failed to connect to all addresses |Gateway: Communication error with deployment at address(es) 0.0.0.0:12346. Head or worker(s) may be down.",
+  "exception": {
+    "name": "InternalNetworkError",
+    "args": [
+      "failed to connect to all addresses |Gateway: Communication error with deployment at address(es) 0.0.0.0:12346. Head or worker(s) may be down."
+    ],
+    "stacks": [
+      "Traceback (most recent call last):\n",
+      "  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 750, in task_wrapper\n    timeout=timeout,\n",
+      "  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 197, in send_discover_endpoint\n    await self._init_stubs()\n",
+      "  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 174, in _init_stubs\n    self.channel\n",
+      "  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 1001, in get_available_services\n    async for res in response:\n",
+      "  File \"/home/joan/.local/lib/python3.7/site-packages/grpc/aio/_call.py\", line 326, in _fetch_stream_responses\n    await self._raise_for_status()\n",
+      "  File \"/home/joan/.local/lib/python3.7/site-packages/grpc/aio/_call.py\", line 237, in _raise_for_status\n    self._cython_call.status())\n",
+      "grpc.aio._call.AioRpcError: \u003cAioRpcError of RPC that terminated with:\n\tstatus = StatusCode.UNAVAILABLE\n\tdetails = \"failed to connect to all addresses\"\n\tdebug_error_string = \"{\"created\":\"@1654012804.794351252\",\"description\":\"Failed to pick subchannel\",\"file\":\"src/core/ext/filters/client_channel/client_channel.cc\",\"file_line\":3134,\"referenced_errors\":[{\"created\":\"@1654012804.794350006\",\"description\":\"failed to connect to all addresses\",\"file\":\"src/core/lib/transport/error_utils.cc\",\"file_line\":163,\"grpc_status\":14}]}\"\n\u003e\n",
+      "\nDuring handling of the above exception, another exception occurred:\n\n",
+      "Traceback (most recent call last):\n",
+      "  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/grpc/__init__.py\", line 155, in dry_run\n    async for _ in self.streamer.stream(request_iterator=req_iterator):\n",
+      "  File \"/home/joan/jina/jina/jina/serve/stream/__init__.py\", line 78, in stream\n    async for response in async_iter:\n",
+      "  File \"/home/joan/jina/jina/jina/serve/stream/__init__.py\", line 154, in _stream_requests\n    response = self._result_handler(future.result())\n",
+      "  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/request_handling.py\", line 146, in _process_results_at_end_gateway\n    await asyncio.gather(gather_endpoints(request_graph))\n",
+      "  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/request_handling.py\", line 88, in gather_endpoints\n    raise err\n",
+      "  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/request_handling.py\", line 80, in gather_endpoints\n    endpoints = await asyncio.gather(*tasks_to_get_endpoints)\n",
+      "  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 754, in task_wrapper\n    e=e, retry_i=i, dest_addr=connection.address\n",
+      "  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 697, in _handle_aiorpcerror\n    details=e.details(),\n",
+      "jina.excepts.InternalNetworkError: failed to connect to all addresses |Gateway: Communication error with deployment at address(es) 0.0.0.0:12346. Head or worker(s) may be down.\n"
+    ]
+  }
+}
+```
+
+
+## With http or websocket
+
+To understand better what is going on under the hood, we can check how to target the endpoint from outside Jina or the client.
+
+Let's first instantiate the Flow and block it for serving.
+
+Then we can use curl to target the `/dry_run` endpoint.
+
+
+```shell
+curl http://localhost:12345/dry_run
+```
+```text
+{"code":0,"description":"","exception":null}%
+```
+
+Let's simulate an Executor going offline by killing its process.
+
+```shell script
+kill -9 $EXECUTOR_PID # in this case we can see in the logs that it is 19059
+```
+
+Then by doing the same check, we will see that it returns an error.
+
+```text
+{"code":1,"description":"failed to connect to all addresses |Gateway: Communication error with deployment executor0 at address(es) {'0.0.0.0:12346'}. Head or worker(s) may be down.","exception":{"name":"InternalNetworkError","args":["failed to connect to all addresses |Gateway: Communication error with deployment executor0 at address(es) {'0.0.0.0:12346'}. Head or worker(s) may be down."],"stacks":["Traceback (most recent call last):\n","  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 726, in task_wrapper\n    timeout=timeout,\n","  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 241, in send_requests\n    await call_result,\n","  File \"/home/joan/.local/lib/python3.7/site-packages/grpc/aio/_call.py\", line 291, in __await__\n    self._cython_call._status)\n","grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:\n\tstatus = StatusCode.UNAVAILABLE\n\tdetails = \"failed to connect to all addresses\"\n\tdebug_error_string = \"{\"created\":\"@1654074272.702044542\",\"description\":\"Failed to pick subchannel\",\"file\":\"src/core/ext/filters/client_channel/client_channel.cc\",\"file_line\":3134,\"referenced_errors\":[{\"created\":\"@1654074272.702043378\",\"description\":\"failed to connect to all addresses\",\"file\":\"src/core/lib/transport/error_utils.cc\",\"file_line\":163,\"grpc_status\":14}]}\"\n>\n","\nDuring handling of the above exception, another exception occurred:\n\n","Traceback (most recent call last):\n","  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/http/app.py\", line 142, in _flow_health\n    data_type=DataInputType.DOCUMENT,\n","  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/http/app.py\", line 399, in _get_singleton_result\n    async for k in streamer.stream(request_iterator=request_iterator):\n","  File \"/home/joan/jina/jina/jina/serve/stream/__init__.py\", line 78, in stream\n    async for response in async_iter:\n","  File \"/home/joan/jina/jina/jina/serve/stream/__init__.py\", line 154, in _stream_requests\n    response = self._result_handler(future.result())\n","  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/request_handling.py\", line 148, in _process_results_at_end_gateway\n    partial_responses = await asyncio.gather(*tasks)\n","  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/graph/topology_graph.py\", line 128, in _wait_previous_and_send\n    self._handle_internalnetworkerror(err)\n","  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/graph/topology_graph.py\", line 70, in _handle_internalnetworkerror\n    raise err\n","  File \"/home/joan/jina/jina/jina/serve/runtimes/gateway/graph/topology_graph.py\", line 125, in _wait_previous_and_send\n    timeout=self._timeout_send,\n","  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 734, in task_wrapper\n    num_retries=num_retries,\n","  File \"/home/joan/jina/jina/jina/serve/networking.py\", line 697, in _handle_aiorpcerror\n    details=e.details(),\n","jina.excepts.InternalNetworkError: failed to connect to all addresses |Gateway: Communication error with deployment executor0 at address(es) {'0.0.0.0:12346'}. Head or worker(s) may be down.\n"],"executor":""}}%
+```
diff --git a/docs/index.md b/docs/index.md
index 6a069c3e99069..570e75d414dda 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -103,6 +103,7 @@ fundamentals/architecture-overview
 fundamentals/executor/index
 fundamentals/flow/index
 fundamentals/clean-code
+fundamentals/health-check
 ```
 
 ```{toctree}
diff --git a/jina/__init__.py b/jina/__init__.py
index 101528b81861e..b457b8b11320c 100644
--- a/jina/__init__.py
+++ b/jina/__init__.py
@@ -69,6 +69,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.11'
+
 try:
     __docarray_version__ = _docarray.__version__
 except AttributeError as e:
diff --git a/jina/clients/base/__init__.py b/jina/clients/base/__init__.py
index b4005b54dab9a..7bb8e2372498b 100644
--- a/jina/clients/base/__init__.py
+++ b/jina/clients/base/__init__.py
@@ -152,6 +152,14 @@ async def _get_results(
     ):
         ...
 
+    @abc.abstractmethod
+    def _dry_run(self, **kwargs) -> bool:
+        """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
+
+        :param kwargs: potential kwargs received passed from the public interface
+        """
+        ...
+
     @property
     def client(self: T) -> T:
         """Return the client object itself
diff --git a/jina/clients/base/grpc.py b/jina/clients/base/grpc.py
index 65097b611d782..21e08223c2303 100644
--- a/jina/clients/base/grpc.py
+++ b/jina/clients/base/grpc.py
@@ -12,7 +12,7 @@
     InternalNetworkError,
 )
 from jina.logging.profile import ProgressBar
-from jina.proto import jina_pb2_grpc
+from jina.proto import jina_pb2, jina_pb2_grpc
 from jina.serve.networking import GrpcConnectionPool
 
 if TYPE_CHECKING:
@@ -31,6 +31,34 @@ class GRPCBaseClient(BaseClient):
     It manages the asyncio event loop internally, so all interfaces are synchronous from the outside.
     """
 
+    async def _dry_run(self, **kwargs) -> bool:
+        """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
+
+        :param kwargs: potential kwargs received passed from the public interface
+        :return: boolean indicating the health/readiness of the Flow
+        """
+        try:
+            async with GrpcConnectionPool.get_grpc_channel(
+                f'{self.args.host}:{self.args.port}',
+                asyncio=True,
+                tls=self.args.tls,
+            ) as channel:
+                stub = jina_pb2_grpc.JinaGatewayDryRunRPCStub(channel)
+                self.logger.debug(f'connected to {self.args.host}:{self.args.port}')
+                call_result = stub.dry_run(
+                    jina_pb2.google_dot_protobuf_dot_empty__pb2.Empty(), **kwargs
+                )
+                metadata, response = (
+                    await call_result.trailing_metadata(),
+                    await call_result,
+                )
+                if response.code == jina_pb2.StatusProto.SUCCESS:
+                    return True
+        except Exception as e:
+            self.logger.error(f'Error while getting response from grpc server {e!r}')
+
+        return False
+
     async def _get_results(
         self,
         inputs: 'InputType',
diff --git a/jina/clients/base/helper.py b/jina/clients/base/helper.py
index a55bb635270bc..6ae060052a8c1 100644
--- a/jina/clients/base/helper.py
+++ b/jina/clients/base/helper.py
@@ -8,6 +8,7 @@
 from jina.importer import ImportExtensions
 from jina.types.request import Request
 from jina.types.request.data import DataRequest
+from jina.types.request.status import StatusMessage
 
 if TYPE_CHECKING:
     from jina.logging.logger import JinaLogger
@@ -33,11 +34,20 @@ async def send_message(self):
         """Send message to Gateway"""
         ...
 
+    @abstractmethod
+    async def send_dry_run(self):
+        """Query the dry_run endpoint from Gateway"""
+        ...
+
     @abstractmethod
     async def recv_message(self):
         """Receive message from Gateway"""
         ...
 
+    async def recv_dry_run(self):
+        """Receive dry run response from Gateway"""
+        pass
+
     async def __aenter__(self):
         """enter async context
 
@@ -83,6 +93,12 @@ async def send_message(self, request: 'Request'):
             req_dict['target_executor'] = req_dict['header']['target_executor']
         return await self.session.post(url=self.url, json=req_dict).__aenter__()
 
+    async def send_dry_run(self):
+        """Query the dry_run endpoint from Gateway
+        :return: send get message
+        """
+        return await self.session.get(url=self.url).__aenter__()
+
     async def recv_message(self):
         """Receive message for HTTP (sleep)
 
@@ -90,6 +106,13 @@ async def recv_message(self):
         """
         return await asyncio.sleep(1e10)
 
+    async def recv_dry_run(self):
+        """Receive dry run response for HTTP (sleep)
+
+        :return: await sleep
+        """
+        return await asyncio.sleep(1e10)
+
 
 class WsResponseIter:
     """
@@ -131,6 +154,17 @@ async def send_message(self, request: 'Request'):
         except ConnectionResetError:
             self.logger.critical(f'server connection closed already!')
 
+    async def send_dry_run(self):
+        """Query the dry_run endpoint from Gateway
+
+        :return: send dry_run as bytes awaitable
+        """
+
+        try:
+            return await self.websocket.send_bytes(b'')
+        except ConnectionResetError:
+            self.logger.critical(f'server connection closed already!')
+
     async def send_eoi(self):
         """To confirm end of iteration, we send `bytes(True)` to the server.
 
@@ -155,6 +189,18 @@ async def recv_message(self) -> 'DataRequest':
         async for response in self.response_iter:
             yield DataRequest(response.data)
 
+    async def recv_dry_run(self):
+        """Receive dry run response in bytes from server
+
+        ..note::
+            aiohttp allows only one task which can `receive` concurrently.
+            we need to make sure we don't create multiple tasks with `recv_message`
+
+        :yield: response objects received from server
+        """
+        async for response in self.response_iter:
+            yield StatusMessage(response.data)
+
     async def __aenter__(self):
         await super().__aenter__()
         self.websocket = await self.session.ws_connect(
diff --git a/jina/clients/base/http.py b/jina/clients/base/http.py
index 56adc19d0018f..57534bc8bb683 100644
--- a/jina/clients/base/http.py
+++ b/jina/clients/base/http.py
@@ -21,6 +21,52 @@
 class HTTPBaseClient(BaseClient):
     """A MixIn for HTTP Client."""
 
+    def _handle_response_status(self, r_status, r_str, url):
+        if r_status == status.HTTP_404_NOT_FOUND:
+            raise BadClient(f'no such endpoint {url}')
+        elif r_status == status.HTTP_503_SERVICE_UNAVAILABLE:
+            if (
+                'header' in r_str
+                and 'status' in r_str['header']
+                and 'description' in r_str['header']['status']
+            ):
+                raise ConnectionError(r_str['header']['status']['description'])
+            else:
+                raise ValueError(r_str)
+        elif (
+            r_status < status.HTTP_200_OK or r_status > status.HTTP_300_MULTIPLE_CHOICES
+        ):  # failure codes
+            raise ValueError(r_str)
+
+    async def _dry_run(self, **kwargs) -> bool:
+        """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
+
+        :param kwargs: potential kwargs received passed from the public interface
+        :return: boolean indicating the health/readiness of the Flow
+        """
+        from jina.proto import jina_pb2
+
+        async with AsyncExitStack() as stack:
+            try:
+                proto = 'https' if self.args.tls else 'http'
+                url = f'{proto}://{self.args.host}:{self.args.port}/dry_run'
+                iolet = await stack.enter_async_context(
+                    HTTPClientlet(url=url, logger=self.logger)
+                )
+
+                response = await iolet.send_dry_run()
+                r_status = response.status
+
+                r_str = await response.json()
+                self._handle_response_status(r_status, r_str, url)
+                if r_str['code'] == jina_pb2.StatusProto.SUCCESS:
+                    return True
+            except Exception as e:
+                self.logger.error(
+                    f'Error while fetching response from HTTP server {e!r}'
+                )
+        return False
+
     async def _get_results(
         self,
         inputs: 'InputType',
@@ -77,24 +123,7 @@ def _result_handler(result):
                     r_status = response.status
 
                     r_str = await response.json()
-                    if r_status == status.HTTP_404_NOT_FOUND:
-                        raise BadClient(f'no such endpoint {url}')
-                    elif r_status == status.HTTP_503_SERVICE_UNAVAILABLE:
-                        if (
-                            'header' in r_str
-                            and 'status' in r_str['header']
-                            and 'description' in r_str['header']['status']
-                        ):
-                            raise ConnectionError(
-                                r_str['header']['status']['description']
-                            )
-                        else:
-                            raise ValueError(r_str)
-                    elif (
-                        r_status < status.HTTP_200_OK
-                        or r_status > status.HTTP_300_MULTIPLE_CHOICES
-                    ):  # failure codes
-                        raise ValueError(r_str)
+                    self._handle_response_status(r_status, r_str, url)
 
                     da = None
                     if 'data' in r_str and r_str['data'] is not None:
diff --git a/jina/clients/base/websocket.py b/jina/clients/base/websocket.py
index 8f5cad8407596..ad8e019b41ee5 100644
--- a/jina/clients/base/websocket.py
+++ b/jina/clients/base/websocket.py
@@ -11,6 +11,7 @@
 from jina.helper import get_or_reuse_loop
 from jina.importer import ImportExtensions
 from jina.logging.profile import ProgressBar
+from jina.proto import jina_pb2
 from jina.serve.stream import RequestStreamer
 
 if TYPE_CHECKING:
@@ -21,6 +22,55 @@
 class WebSocketBaseClient(BaseClient):
     """A Websocket Client."""
 
+    async def _dry_run(self, **kwargs) -> bool:
+        """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
+
+        :param kwargs: potential kwargs received passed from the public interface
+        :return: boolean indicating the readiness of the Flow
+        """
+        async with AsyncExitStack() as stack:
+            try:
+                proto = 'wss' if self.args.tls else 'ws'
+                url = f'{proto}://{self.args.host}:{self.args.port}/dry_run'
+                iolet = await stack.enter_async_context(
+                    WebsocketClientlet(url=url, logger=self.logger)
+                )
+
+                async def _receive():
+                    try:
+                        async for response in iolet.recv_dry_run():
+                            return response
+                    except Exception as exc:
+                        self.logger.error(
+                            f'Error while fetching response from Websocket server {exc!r}'
+                        )
+                        raise
+
+                async def _send():
+                    return await iolet.send_dry_run()
+
+                receive_task = asyncio.create_task(_receive())
+
+                if receive_task.done():
+                    raise RuntimeError(
+                        'receive task not running, can not send messages'
+                    )
+                try:
+                    send_task = asyncio.create_task(_send())
+                    _, response_result = await asyncio.gather(send_task, receive_task)
+                    if response_result.proto.code == jina_pb2.StatusProto.SUCCESS:
+                        return True
+                finally:
+                    if iolet.close_code == status.WS_1011_INTERNAL_ERROR:
+                        raise ConnectionError(iolet.close_message)
+                    await receive_task
+
+            except Exception as e:
+                self.logger.error(
+                    f'Error while getting response from websocket server {e!r}'
+                )
+            return False
+
     async def _get_results(
         self,
         inputs: 'InputType',
diff --git a/jina/clients/grpc.py b/jina/clients/grpc.py
index 9b18ef4dc5fdf..2bca0d5c5ccb9 100644
--- a/jina/clients/grpc.py
+++ b/jina/clients/grpc.py
@@ -1,12 +1,12 @@
 from jina.clients.base.grpc import GRPCBaseClient
-from jina.clients.mixin import AsyncPostMixin, PostMixin
+from jina.clients.mixin import AsyncPostMixin, HealthCheckMixin, PostMixin
 
 
-class GRPCClient(GRPCBaseClient, PostMixin):
+class GRPCClient(GRPCBaseClient, PostMixin, HealthCheckMixin):
     """A client communicates the server with GRPC protocol."""
 
 
-class AsyncGRPCClient(GRPCBaseClient, AsyncPostMixin):
+class AsyncGRPCClient(GRPCBaseClient, AsyncPostMixin, HealthCheckMixin):
     """
     A client communicates the server with GRPC protocol.
 
diff --git a/jina/clients/http.py b/jina/clients/http.py
index dba662ac07f3a..ae1023ea899d7 100644
--- a/jina/clients/http.py
+++ b/jina/clients/http.py
@@ -1,14 +1,22 @@
 from jina.clients.base.http import HTTPBaseClient
-from jina.clients.mixin import AsyncPostMixin, PostMixin, MutateMixin, AsyncMutateMixin
+from jina.clients.mixin import (
+    AsyncMutateMixin,
+    AsyncPostMixin,
+    HealthCheckMixin,
+    MutateMixin,
+    PostMixin,
+)
 
 
-class HTTPClient(HTTPBaseClient, PostMixin, MutateMixin):
+class HTTPClient(HTTPBaseClient, PostMixin, MutateMixin, HealthCheckMixin):
     """
     A client communicates the server with HTTP protocol.
     """
 
 
-class AsyncHTTPClient(HTTPBaseClient, AsyncPostMixin, AsyncMutateMixin):
+class AsyncHTTPClient(
+    HTTPBaseClient, AsyncPostMixin, AsyncMutateMixin, HealthCheckMixin
+):
     """
     A client communicates the server with HTTP protocol.
 
diff --git a/jina/clients/mixin.py b/jina/clients/mixin.py
index 21d3e4f75689f..4fa3b6d28040c 100644
--- a/jina/clients/mixin.py
+++ b/jina/clients/mixin.py
@@ -88,6 +88,18 @@ async def mutate(
         )
 
 
+class HealthCheckMixin:
+    """The Health check Mixin for Client and Flow to expose `dry_run` API"""
+
+    def dry_run(self, **kwargs) -> bool:
+        """Sends a dry run to the Flow to validate if the Flow is ready to receive requests
+
+        :param kwargs: potential kwargs received passed from the public interface
+        :return: boolean indicating the health/readiness of the Flow
+        """
+        return run_async(self.client._dry_run, **kwargs)
+
+
 class PostMixin:
     """The Post Mixin class for Client and Flow"""
 
diff --git a/jina/clients/websocket.py b/jina/clients/websocket.py
index c488c4aff42d8..036298e9c0bbf 100644
--- a/jina/clients/websocket.py
+++ b/jina/clients/websocket.py
@@ -1,14 +1,14 @@
 from jina.clients.base.websocket import WebSocketBaseClient
-from jina.clients.mixin import AsyncPostMixin, PostMixin
+from jina.clients.mixin import AsyncPostMixin, HealthCheckMixin, PostMixin
 
 
-class WebSocketClient(WebSocketBaseClient, PostMixin):
+class WebSocketClient(WebSocketBaseClient, PostMixin, HealthCheckMixin):
     """
     A client communicates the server with WebSocket protocol.
     """
 
 
-class AsyncWebSocketClient(WebSocketBaseClient, AsyncPostMixin):
+class AsyncWebSocketClient(WebSocketBaseClient, AsyncPostMixin, HealthCheckMixin):
     """
     A client communicates the server with WebSocket protocol.
 
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index e8047bd728e02..c7f26abec92d2 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -38,7 +38,7 @@
 
 from jina import __default_host__, __default_port_monitoring__, __docker_host__, helper
 from jina.clients import Client
-from jina.clients.mixin import AsyncPostMixin, PostMixin
+from jina.clients.mixin import AsyncPostMixin, HealthCheckMixin, PostMixin
 from jina.enums import (
     DeploymentRoleType,
     FlowBuildLevel,
@@ -97,7 +97,7 @@ class FlowType(type(ExitStack), type(JAMLCompatible)):
 ]
 
 
-class Flow(PostMixin, JAMLCompatible, ExitStack, metaclass=FlowType):
+class Flow(PostMixin, HealthCheckMixin, JAMLCompatible, ExitStack, metaclass=FlowType):
     """Flow is how Jina streamlines and distributes Executors."""
 
     # overload_inject_start_client_flow
diff --git a/jina/proto/jina.proto b/jina/proto/jina.proto
index f97cefc5cccc9..5689cde828318 100644
--- a/jina/proto/jina.proto
+++ b/jina/proto/jina.proto
@@ -160,3 +160,12 @@ service JinaDiscoverEndpointsRPC {
     }
 }
 
+
+/**
+ * jina gRPC service to expose Endpoints from Executors.
+ */
+service JinaGatewayDryRunRPC {
+    rpc dry_run (google.protobuf.Empty) returns (StatusProto) {
+    }
+}
+
diff --git a/jina/proto/jina_pb2.py b/jina/proto/jina_pb2.py
index 82b82ff5a838c..d0879789e957b 100644
--- a/jina/proto/jina_pb2.py
+++ b/jina/proto/jina_pb2.py
@@ -18,7 +18,7 @@
 from google.protobuf import struct_pb2 as google_dot_protobuf_dot_struct__pb2
 from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2
 
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\njina.proto\x12\x04jina\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1cgoogle/protobuf/struct.proto\x1a\x1bgoogle/protobuf/empty.proto\x1a\x0e\x64ocarray.proto\"\x9f\x01\n\nRouteProto\x12\x10\n\x08\x65xecutor\x18\x01 \x01(\t\x12.\n\nstart_time\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12,\n\x08\x65nd_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12!\n\x06status\x18\x04 \x01(\x0b\x32\x11.jina.StatusProto\"\xc6\x01\n\x0bHeaderProto\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12!\n\x06status\x18\x02 \x01(\x0b\x32\x11.jina.StatusProto\x12\x1a\n\rexec_endpoint\x18\x03 \x01(\tH\x00\x88\x01\x01\x12\x1c\n\x0ftarget_executor\x18\x04 \x01(\tH\x01\x88\x01\x01\x12\x14\n\x07timeout\x18\x05 \x01(\rH\x02\x88\x01\x01\x42\x10\n\x0e_exec_endpointB\x12\n\x10_target_executorB\n\n\x08_timeout\"#\n\x0e\x45ndpointsProto\x12\x11\n\tendpoints\x18\x01 \x03(\t\"\xf9\x01\n\x0bStatusProto\x12*\n\x04\x63ode\x18\x01 \x01(\x0e\x32\x1c.jina.StatusProto.StatusCode\x12\x13\n\x0b\x64\x65scription\x18\x02 \x01(\t\x12\x33\n\texception\x18\x03 \x01(\x0b\x32 .jina.StatusProto.ExceptionProto\x1aN\n\x0e\x45xceptionProto\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0c\n\x04\x61rgs\x18\x02 \x03(\t\x12\x0e\n\x06stacks\x18\x03 \x03(\t\x12\x10\n\x08\x65xecutor\x18\x04 \x01(\t\"$\n\nStatusCode\x12\x0b\n\x07SUCCESS\x10\x00\x12\t\n\x05\x45RROR\x10\x01\"^\n\rRelatedEntity\x12\n\n\x02id\x18\x01 \x01(\t\x12\x0f\n\x07\x61\x64\x64ress\x18\x02 \x01(\t\x12\x0c\n\x04port\x18\x03 \x01(\r\x12\x15\n\x08shard_id\x18\x04 \x01(\rH\x00\x88\x01\x01\x42\x0b\n\t_shard_id\"\xa0\x02\n\x10\x44\x61taRequestProto\x12!\n\x06header\x18\x01 \x01(\x0b\x32\x11.jina.HeaderProto\x12+\n\nparameters\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\x12 \n\x06routes\x18\x03 \x03(\x0b\x32\x10.jina.RouteProto\x12\x35\n\x04\x64\x61ta\x18\x04 \x01(\x0b\x32\'.jina.DataRequestProto.DataContentProto\x1a\x63\n\x10\x44\x61taContentProto\x12,\n\x04\x64ocs\x18\x01 \x01(\x0b\x32\x1c.docarray.DocumentArrayProtoH\x00\x12\x14\n\ndocs_bytes\x18\x02 \x01(\x0cH\x00\x42\x0b\n\tdocuments\"@\n\x14\x44\x61taRequestListProto\x12(\n\x08requests\x18\x01 \x03(\x0b\x32\x16.jina.DataRequestProto2Z\n\x12JinaDataRequestRPC\x12\x44\n\x0cprocess_data\x12\x1a.jina.DataRequestListProto\x1a\x16.jina.DataRequestProto\"\x00\x32\x63\n\x18JinaSingleDataRequestRPC\x12G\n\x13process_single_data\x12\x16.jina.DataRequestProto\x1a\x16.jina.DataRequestProto\"\x00\x32G\n\x07JinaRPC\x12<\n\x04\x43\x61ll\x12\x16.jina.DataRequestProto\x1a\x16.jina.DataRequestProto\"\x00(\x01\x30\x01\x32`\n\x18JinaDiscoverEndpointsRPC\x12\x44\n\x12\x65ndpoint_discovery\x12\x16.google.protobuf.Empty\x1a\x14.jina.EndpointsProto\"\x00\x62\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\njina.proto\x12\x04jina\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1cgoogle/protobuf/struct.proto\x1a\x1bgoogle/protobuf/empty.proto\x1a\x0e\x64ocarray.proto\"\x9f\x01\n\nRouteProto\x12\x10\n\x08\x65xecutor\x18\x01 \x01(\t\x12.\n\nstart_time\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12,\n\x08\x65nd_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12!\n\x06status\x18\x04 \x01(\x0b\x32\x11.jina.StatusProto\"\xc6\x01\n\x0bHeaderProto\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12!\n\x06status\x18\x02 \x01(\x0b\x32\x11.jina.StatusProto\x12\x1a\n\rexec_endpoint\x18\x03 \x01(\tH\x00\x88\x01\x01\x12\x1c\n\x0ftarget_executor\x18\x04 \x01(\tH\x01\x88\x01\x01\x12\x14\n\x07timeout\x18\x05 \x01(\rH\x02\x88\x01\x01\x42\x10\n\x0e_exec_endpointB\x12\n\x10_target_executorB\n\n\x08_timeout\"#\n\x0e\x45ndpointsProto\x12\x11\n\tendpoints\x18\x01 \x03(\t\"\xf9\x01\n\x0bStatusProto\x12*\n\x04\x63ode\x18\x01 \x01(\x0e\x32\x1c.jina.StatusProto.StatusCode\x12\x13\n\x0b\x64\x65scription\x18\x02 \x01(\t\x12\x33\n\texception\x18\x03 \x01(\x0b\x32 .jina.StatusProto.ExceptionProto\x1aN\n\x0e\x45xceptionProto\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0c\n\x04\x61rgs\x18\x02 \x03(\t\x12\x0e\n\x06stacks\x18\x03 \x03(\t\x12\x10\n\x08\x65xecutor\x18\x04 \x01(\t\"$\n\nStatusCode\x12\x0b\n\x07SUCCESS\x10\x00\x12\t\n\x05\x45RROR\x10\x01\"^\n\rRelatedEntity\x12\n\n\x02id\x18\x01 \x01(\t\x12\x0f\n\x07\x61\x64\x64ress\x18\x02 \x01(\t\x12\x0c\n\x04port\x18\x03 \x01(\r\x12\x15\n\x08shard_id\x18\x04 \x01(\rH\x00\x88\x01\x01\x42\x0b\n\t_shard_id\"\xa0\x02\n\x10\x44\x61taRequestProto\x12!\n\x06header\x18\x01 \x01(\x0b\x32\x11.jina.HeaderProto\x12+\n\nparameters\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\x12 \n\x06routes\x18\x03 \x03(\x0b\x32\x10.jina.RouteProto\x12\x35\n\x04\x64\x61ta\x18\x04 \x01(\x0b\x32\'.jina.DataRequestProto.DataContentProto\x1a\x63\n\x10\x44\x61taContentProto\x12,\n\x04\x64ocs\x18\x01 \x01(\x0b\x32\x1c.docarray.DocumentArrayProtoH\x00\x12\x14\n\ndocs_bytes\x18\x02 \x01(\x0cH\x00\x42\x0b\n\tdocuments\"@\n\x14\x44\x61taRequestListProto\x12(\n\x08requests\x18\x01 \x03(\x0b\x32\x16.jina.DataRequestProto2Z\n\x12JinaDataRequestRPC\x12\x44\n\x0cprocess_data\x12\x1a.jina.DataRequestListProto\x1a\x16.jina.DataRequestProto\"\x00\x32\x63\n\x18JinaSingleDataRequestRPC\x12G\n\x13process_single_data\x12\x16.jina.DataRequestProto\x1a\x16.jina.DataRequestProto\"\x00\x32G\n\x07JinaRPC\x12<\n\x04\x43\x61ll\x12\x16.jina.DataRequestProto\x1a\x16.jina.DataRequestProto\"\x00(\x01\x30\x01\x32`\n\x18JinaDiscoverEndpointsRPC\x12\x44\n\x12\x65ndpoint_discovery\x12\x16.google.protobuf.Empty\x1a\x14.jina.EndpointsProto\"\x00\x32N\n\x14JinaGatewayDryRunRPC\x12\x36\n\x07\x64ry_run\x12\x16.google.protobuf.Empty\x1a\x11.jina.StatusProto\"\x00\x62\x06proto3')
 
 
 
@@ -101,6 +101,7 @@
 _JINASINGLEDATAREQUESTRPC = DESCRIPTOR.services_by_name['JinaSingleDataRequestRPC']
 _JINARPC = DESCRIPTOR.services_by_name['JinaRPC']
 _JINADISCOVERENDPOINTSRPC = DESCRIPTOR.services_by_name['JinaDiscoverEndpointsRPC']
+_JINAGATEWAYDRYRUNRPC = DESCRIPTOR.services_by_name['JinaGatewayDryRunRPC']
 if _descriptor._USE_C_DESCRIPTORS == False:
 
   DESCRIPTOR._options = None
@@ -132,4 +133,6 @@
   _JINARPC._serialized_end=1497
   _JINADISCOVERENDPOINTSRPC._serialized_start=1499
   _JINADISCOVERENDPOINTSRPC._serialized_end=1595
+  _JINAGATEWAYDRYRUNRPC._serialized_start=1597
+  _JINAGATEWAYDRYRUNRPC._serialized_end=1675
 # @@protoc_insertion_point(module_scope)
diff --git a/jina/proto/jina_pb2_grpc.py b/jina/proto/jina_pb2_grpc.py
index c632e18a3ba1a..4575da9a47ba5 100644
--- a/jina/proto/jina_pb2_grpc.py
+++ b/jina/proto/jina_pb2_grpc.py
@@ -327,3 +327,83 @@ def endpoint_discovery(
             timeout,
             metadata,
         )
+
+
+class JinaGatewayDryRunRPCStub(object):
+    """*
+    jina gRPC service to expose Endpoints from Executors.
+    """
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.dry_run = channel.unary_unary(
+            '/jina.JinaGatewayDryRunRPC/dry_run',
+            request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString,
+            response_deserializer=jina__pb2.StatusProto.FromString,
+        )
+
+
+class JinaGatewayDryRunRPCServicer(object):
+    """*
+    jina gRPC service to expose Endpoints from Executors.
+    """
+
+    def dry_run(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_JinaGatewayDryRunRPCServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+        'dry_run': grpc.unary_unary_rpc_method_handler(
+            servicer.dry_run,
+            request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString,
+            response_serializer=jina__pb2.StatusProto.SerializeToString,
+        ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+        'jina.JinaGatewayDryRunRPC', rpc_method_handlers
+    )
+    server.add_generic_rpc_handlers((generic_handler,))
+
+
+# This class is part of an EXPERIMENTAL API.
+class JinaGatewayDryRunRPC(object):
+    """*
+    jina gRPC service to expose Endpoints from Executors.
+    """
+
+    @staticmethod
+    def dry_run(
+        request,
+        target,
+        options=(),
+        channel_credentials=None,
+        call_credentials=None,
+        insecure=False,
+        compression=None,
+        wait_for_ready=None,
+        timeout=None,
+        metadata=None,
+    ):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/jina.JinaGatewayDryRunRPC/dry_run',
+            google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString,
+            jina__pb2.StatusProto.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+        )
diff --git a/jina/proto/serializer.py b/jina/proto/serializer.py
index 045542a55817c..bbdf0d9f796f5 100644
--- a/jina/proto/serializer.py
+++ b/jina/proto/serializer.py
@@ -100,3 +100,30 @@ def FromString(x: bytes):
         ep.ParseFromString(x)
 
         return ep
+
+
+class StatusProto:
+    """Since the serializer is replacing the `jina_pb2 to know how to exactly serialize messages, this is just a placeholder that
+    delegates the serializing and deserializing to the internal protobuf structure with no extra optimization.
+    """
+
+    @staticmethod
+    def SerializeToString(x):
+        """
+        # noqa: DAR101
+        # noqa: DAR102
+        # noqa: DAR201
+        """
+        return x.SerializeToString()
+
+    @staticmethod
+    def FromString(x: bytes):
+        """
+        # noqa: DAR101
+        # noqa: DAR102
+        # noqa: DAR201
+        """
+        sp = jina_pb2.StatusProto()
+        sp.ParseFromString(x)
+
+        return sp
diff --git a/jina/serve/executors/__init__.py b/jina/serve/executors/__init__.py
index 92c5a96ee5fb3..f3d2c390504e5 100644
--- a/jina/serve/executors/__init__.py
+++ b/jina/serve/executors/__init__.py
@@ -20,7 +20,9 @@
 
     from docarray import DocumentArray
 
-__all__ = ['BaseExecutor', 'ReducerExecutor']
+__dry_run_endpoint__ = '_jina_dry_run_'
+
+__all__ = ['BaseExecutor', 'ReducerExecutor', __dry_run_endpoint__]
 
 
 class ExecutorType(type(JAMLCompatible), type):
@@ -110,6 +112,16 @@ def __init__(
         self._add_runtime_args(runtime_args)
         self._init_monitoring()
         self.logger = JinaLogger(self.__class__.__name__)
+        if __dry_run_endpoint__ not in self.requests:
+            self.requests[__dry_run_endpoint__] = self._dry_run_func
+        else:
+            self.logger.warning(
+                f' Endpoint {__dry_run_endpoint__} is defined by the Executor. Be aware that this endpoint is usually reserved to enable health checks from the Client through the gateway.'
+                f' So it is recommended not to expose this endpoint. '
+            )
+
+    def _dry_run_func(self, *args, **kwargs):
+        pass
 
     def _add_runtime_args(self, _runtime_args: Optional[Dict]):
         if _runtime_args:
diff --git a/jina/serve/networking.py b/jina/serve/networking.py
index 0dbd418103268..11314b715bc0d 100644
--- a/jina/serve/networking.py
+++ b/jina/serve/networking.py
@@ -1003,7 +1003,11 @@ async def get_available_services(channel) -> List[str]:
                 [
                     service.name
                     for service in res.list_services_response.service
-                    if service.name != 'grpc.reflection.v1alpha.ServerReflection'
+                    if service.name
+                    not in {
+                        'grpc.reflection.v1alpha.ServerReflection',
+                        'jina.JinaGatewayDryRunRPC',
+                    }
                 ]
             )
         return service_names[0]
diff --git a/jina/serve/runtimes/gateway/graph/topology_graph.py b/jina/serve/runtimes/gateway/graph/topology_graph.py
index c1c4dc8e6b6cc..3df86cdcf67f9 100644
--- a/jina/serve/runtimes/gateway/graph/topology_graph.py
+++ b/jina/serve/runtimes/gateway/graph/topology_graph.py
@@ -71,7 +71,7 @@ def _handle_internalnetworkerror(self, err):
             else:
                 raise
 
-        def get_endpoints(self, connection_pool: GrpcConnectionPool):
+        def get_endpoints(self, connection_pool: GrpcConnectionPool) -> asyncio.Task:
             return connection_pool.send_discover_endpoint(self.name)
 
         async def _wait_previous_and_send(
@@ -300,3 +300,29 @@ def origin_nodes(self):
         :return: A list of nodes
         """
         return self._origin_nodes
+
+    @property
+    def all_nodes(self):
+        """
+        The set of all the nodes inside this Graph
+
+        :return: A list of nodes
+        """
+
+        def _get_all_nodes(node, accum, accum_names):
+            if node.name not in accum_names:
+                accum.append(node)
+                accum_names.append(node.name)
+            for n in node.outgoing_nodes:
+                _get_all_nodes(n, accum, accum_names)
+            return accum, accum_names
+
+        nodes = []
+        node_names = []
+        for origin_node in self.origin_nodes:
+            subtree_nodes, subtree_node_names = _get_all_nodes(origin_node, [], [])
+            for st_node, st_node_name in zip(subtree_nodes, subtree_node_names):
+                if st_node_name not in node_names:
+                    nodes.append(st_node)
+                    node_names.append(st_node_name)
+        return nodes
diff --git a/jina/serve/runtimes/gateway/grpc/__init__.py b/jina/serve/runtimes/gateway/grpc/__init__.py
index 9d6453a72b838..0fc8cf5b0da18 100644
--- a/jina/serve/runtimes/gateway/grpc/__init__.py
+++ b/jina/serve/runtimes/gateway/grpc/__init__.py
@@ -12,6 +12,7 @@
 from jina.serve.runtimes.gateway import GatewayRuntime
 from jina.serve.runtimes.gateway.request_handling import RequestHandler
 from jina.serve.stream import RequestStreamer
+from jina.types.request.status import StatusMessage
 
 __all__ = ['GRPCGatewayRuntime']
 
@@ -71,9 +72,11 @@ async def _async_setup_server(self):
         self.streamer.Call = self.streamer.stream
 
         jina_pb2_grpc.add_JinaRPCServicer_to_server(self.streamer, self.server)
+        jina_pb2_grpc.add_JinaGatewayDryRunRPCServicer_to_server(self, self.server)
 
         service_names = (
             jina_pb2.DESCRIPTOR.services_by_name['JinaRPC'].full_name,
+            jina_pb2.DESCRIPTOR.services_by_name['JinaGatewayDryRunRPC'].full_name,
             reflection.SERVICE_NAME,
         )
         # Mark all services as healthy.
@@ -127,3 +130,34 @@ async def async_run_forever(self):
         """The async running of server."""
         self._connection_pool.start()
         await self.server.wait_for_termination()
+
+    async def dry_run(self, empty, context) -> jina_pb2.StatusProto:
+        """
+        Process the the call requested by having a dry run call to every Executor in the graph
+
+        :param empty: The service expects an empty protobuf message
+        :param context: grpc context
+        :returns: the response request
+        """
+        from docarray import DocumentArray
+        from jina.clients.request import request_generator
+        from jina.enums import DataInputType
+        from jina.serve.executors import __dry_run_endpoint__
+
+        da = DocumentArray()
+
+        try:
+            req_iterator = request_generator(
+                exec_endpoint=__dry_run_endpoint__,
+                data=da,
+                data_type=DataInputType.DOCUMENT,
+            )
+            async for _ in self.streamer.stream(request_iterator=req_iterator):
+                pass
+            status_message = StatusMessage()
+            status_message.set_code(jina_pb2.StatusProto.SUCCESS)
+            return status_message.proto
+        except Exception as ex:
+            status_message = StatusMessage()
+            status_message.set_exception(ex)
+            return status_message.proto
diff --git a/jina/serve/runtimes/gateway/http/app.py b/jina/serve/runtimes/gateway/http/app.py
index f51e5fc65e7cf..b7af83f7ef8dc 100644
--- a/jina/serve/runtimes/gateway/http/app.py
+++ b/jina/serve/runtimes/gateway/http/app.py
@@ -102,17 +102,54 @@ async def _shutdown():
 
         @app.get(
             path='/',
-            summary='Get the health of Jina service',
+            summary='Get the health of Jina Gateway service',
             response_model=JinaHealthModel,
         )
-        async def _health():
+        async def _gateway_health():
             """
-            Get the health of this Jina service.
+            Get the health of this Gateway service.
             .. # noqa: DAR201
 
             """
             return {}
 
+        from docarray import DocumentArray
+        from jina.proto import jina_pb2
+        from jina.serve.executors import __dry_run_endpoint__
+        from jina.serve.runtimes.gateway.http.models import PROTO_TO_PYDANTIC_MODELS
+        from jina.types.request.status import StatusMessage
+
+        @app.get(
+            path='/dry_run',
+            summary='Get the readiness of Jina Flow service, sends an empty DocumentArray to the complete Flow to '
+            'validate connectivity',
+            response_model=PROTO_TO_PYDANTIC_MODELS.StatusProto,
+        )
+        async def _flow_health():
+            """
+            Get the health of the complete Flow service.
+            .. # noqa: DAR201
+
+            """
+
+            da = DocumentArray()
+
+            try:
+                _ = await _get_singleton_result(
+                    request_generator(
+                        exec_endpoint=__dry_run_endpoint__,
+                        data=da,
+                        data_type=DataInputType.DOCUMENT,
+                    )
+                )
+                status_message = StatusMessage()
+                status_message.set_code(jina_pb2.StatusProto.SUCCESS)
+                return status_message.to_dict()
+            except Exception as ex:
+                status_message = StatusMessage()
+                status_message.set_exception(ex)
+                return status_message.to_dict(use_integers_for_enums=True)
+
         @app.get(
             path='/status',
             summary='Get the status of Jina service',
diff --git a/jina/serve/runtimes/gateway/request_handling.py b/jina/serve/runtimes/gateway/request_handling.py
index 804ce7421e708..bc261ff6e2701 100644
--- a/jina/serve/runtimes/gateway/request_handling.py
+++ b/jina/serve/runtimes/gateway/request_handling.py
@@ -4,8 +4,8 @@
 from typing import TYPE_CHECKING, Callable, List, Optional
 
 import grpc.aio
-from docarray import DocumentArray
 
+from docarray import DocumentArray
 from jina.excepts import InternalNetworkError
 from jina.importer import ImportExtensions
 from jina.serve.networking import GrpcConnectionPool
@@ -72,22 +72,7 @@ def handle_request(
         """
 
         async def gather_endpoints(request_graph):
-            def _get_all_nodes(node, accum, accum_names):
-                if node.name not in accum_names:
-                    accum.append(node)
-                    accum_names.append(node.name)
-                for n in node.outgoing_nodes:
-                    _get_all_nodes(n, accum, accum_names)
-                return accum, accum_names
-
-            nodes = []
-            node_names = []
-            for origin_node in request_graph.origin_nodes:
-                subtree_nodes, subtree_node_names = _get_all_nodes(origin_node, [], [])
-                for st_node, st_node_name in zip(subtree_nodes, subtree_node_names):
-                    if st_node_name not in node_names:
-                        nodes.append(st_node)
-                        node_names.append(st_node_name)
+            nodes = request_graph.all_nodes
             try:
                 tasks_to_get_endpoints = [
                     node.get_endpoints(connection_pool) for node in nodes
diff --git a/jina/serve/runtimes/gateway/websocket/app.py b/jina/serve/runtimes/gateway/websocket/app.py
index 61366c8232235..0126039849297 100644
--- a/jina/serve/runtimes/gateway/websocket/app.py
+++ b/jina/serve/runtimes/gateway/websocket/app.py
@@ -1,13 +1,14 @@
 import argparse
-import json
-from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Union
 
+from docarray import DocumentArray
 from jina.clients.request import request_generator
 from jina.enums import DataInputType, WebsocketSubProtocols
 from jina.excepts import InternalNetworkError
 from jina.importer import ImportExtensions
 from jina.logging.logger import JinaLogger
 from jina.types.request.data import DataRequest
+from jina.types.request.status import StatusMessage
 
 if TYPE_CHECKING:
     from prometheus_client import CollectorRegistry
@@ -104,7 +105,9 @@ async def iter(self, websocket: WebSocket) -> AsyncIterator[Any]:
             except WebSocketDisconnect:
                 pass
 
-        async def send(self, websocket: WebSocket, data: DataRequest) -> None:
+        async def send(
+            self, websocket: WebSocket, data: Union[DataRequest, StatusMessage]
+        ) -> None:
             subprotocol = self.protocol_dict[self.get_client(websocket)]
             if subprotocol == WebsocketSubProtocols.JSON:
                 return await websocket.send_json(data.to_dict(), mode='text')
@@ -190,4 +193,90 @@ async def req_iter():
             logger.info('Client successfully disconnected from server')
             manager.disconnect(websocket)
 
+    async def _get_singleton_result(request_iterator) -> Dict:
+        """
+        Streams results from AsyncPrefetchCall as a dict
+
+        :param request_iterator: request iterator, with length of 1
+        :return: the first result from the request iterator
+        """
+        async for k in streamer.stream(request_iterator=request_iterator):
+            request_dict = k.to_dict()
+            return request_dict
+
+    from docarray import DocumentArray
+    from jina.proto import jina_pb2
+    from jina.serve.executors import __dry_run_endpoint__
+    from jina.serve.runtimes.gateway.http.models import PROTO_TO_PYDANTIC_MODELS
+
+    @app.get(
+        path='/dry_run',
+        summary='Get the readiness of Jina Flow service, sends an empty DocumentArray to the complete Flow to '
+        'validate connectivity',
+        response_model=PROTO_TO_PYDANTIC_MODELS.StatusProto,
+    )
+    async def _dry_run_http():
+        """
+        Get the health of the complete Flow service.
+        .. # noqa: DAR201
+
+        """
+
+        da = DocumentArray()
+
+        try:
+            _ = await _get_singleton_result(
+                request_generator(
+                    exec_endpoint=__dry_run_endpoint__,
+                    data=da,
+                    data_type=DataInputType.DOCUMENT,
+                )
+            )
+            status_message = StatusMessage()
+            status_message.set_code(jina_pb2.StatusProto.SUCCESS)
+            return status_message.to_dict()
+        except Exception as ex:
+            status_message = StatusMessage()
+            status_message.set_exception(ex)
+            return status_message.to_dict(use_integers_for_enums=True)
+
+    @app.websocket('/dry_run')
+    async def websocket_endpoint(
+        websocket: WebSocket, response: Response
+    ):  # 'response' is a FastAPI response, not a Jina response
+        from jina.proto import jina_pb2
+        from jina.serve.executors import __dry_run_endpoint__
+
+        await manager.connect(websocket)
+
+        da = DocumentArray()
+        try:
+            async for _ in streamer.stream(
+                request_iterator=request_generator(
+                    exec_endpoint=__dry_run_endpoint__,
+                    data=da,
+                    data_type=DataInputType.DOCUMENT,
+                )
+            ):
+                pass
+            status_message = StatusMessage()
+            status_message.set_code(jina_pb2.StatusProto.SUCCESS)
+            await manager.send(websocket, status_message)
+        except InternalNetworkError as err:
+            manager.disconnect(websocket)
+            msg = (
+                err.details()
+                if _fits_ws_close_msg(err.details())  # some messages are too long
+                else f'Network error while connecting to deployment at {err.dest_addr}. It may be down.'
+            )
+            await websocket.close(code=status.WS_1011_INTERNAL_ERROR, reason=msg)
+        except WebSocketDisconnect:
+            logger.info('Client successfully disconnected from server')
+            manager.disconnect(websocket)
+        except Exception as ex:
+            manager.disconnect(websocket)
+            status_message = StatusMessage()
+            status_message.set_exception(ex)
+            await manager.send(websocket, status_message)
+
     return app
diff --git a/jina/types/mixin.py b/jina/types/mixin.py
index e6961b9d6fbdb..8dff0234a3325 100644
--- a/jina/types/mixin.py
+++ b/jina/types/mixin.py
@@ -1,6 +1,6 @@
 from typing import Dict
 
-from jina.helper import typename, T, TYPE_CHECKING, deprecate_by
+from jina.helper import TYPE_CHECKING, T, deprecate_by, typename
 
 if TYPE_CHECKING:
     from jina.proto import jina_pb2
@@ -18,7 +18,6 @@ class ProtoTypeMixin:
             .. code-block:: python
 
                 class MyJinaType(ProtoTypeMixin):
-
                     def __init__(self, proto: Optional[jina_pb2.SomePbMsg] = None):
                         self._pb_body = proto or jina_pb2.SomePbMsg()
 
@@ -35,21 +34,20 @@ def to_json(self) -> str:
             self.proto, preserving_proto_field_name=True, sort_keys=True
         )
 
-    def to_dict(self) -> Dict:
+    def to_dict(self, **kwargs) -> Dict:
         """Return the object in Python dictionary.
 
         .. note::
             Array like object such as :class:`numpy.ndarray` (i.e. anything described as :class:`jina_pb2.NdArrayProto`)
             will be converted to Python list.
 
+        :param kwargs: Extra kwargs to be passed to MessageToDict, like use_integers_for_enums
+
         :return: dict representation of the object
         """
         from google.protobuf.json_format import MessageToDict
 
-        return MessageToDict(
-            self.proto,
-            preserving_proto_field_name=True,
-        )
+        return MessageToDict(self.proto, preserving_proto_field_name=True, **kwargs)
 
     @property
     def proto(self) -> 'jina_pb2._reflection.GeneratedProtocolMessageType':
diff --git a/jina/types/request/status.py b/jina/types/request/status.py
new file mode 100644
index 0000000000000..fac4312f7f592
--- /dev/null
+++ b/jina/types/request/status.py
@@ -0,0 +1,60 @@
+from typing import Dict, Optional, TypeVar
+
+from google.protobuf import json_format
+
+from jina.excepts import BadRequestType
+from jina.helper import typename
+from jina.proto import jina_pb2
+from jina.types.mixin import ProtoTypeMixin
+
+StatusSourceType = TypeVar('StatusSourceType', jina_pb2.StatusProto, str, Dict, bytes)
+
+
+class StatusMessage(ProtoTypeMixin):
+    """Represents a Status message used for health check of the Flow"""
+
+    def __init__(
+        self,
+        status_object: Optional[StatusSourceType] = None,
+    ):
+        self._pb_body = jina_pb2.StatusProto()
+        try:
+            if isinstance(status_object, jina_pb2.StatusProto):
+                self._pb_body = status_object
+            elif isinstance(status_object, dict):
+                json_format.ParseDict(status_object, self._pb_body)
+            elif isinstance(status_object, str):
+                json_format.Parse(status_object, self._pb_body)
+            elif isinstance(status_object, bytes):
+                self._pb_body.ParseFromString(status_object)
+            elif status_object is not None:
+                # note ``None`` is not considered as a bad type
+                raise ValueError(f'{typename(status_object)} is not recognizable')
+            else:
+                self._pb_body = jina_pb2.StatusProto()
+        except Exception as ex:
+            raise BadRequestType(
+                f'fail to construct a {self.__class__} object from {status_object}'
+            ) from ex
+
+    def set_exception(self, ex: Exception):
+        """Set exception information into the Status Message
+
+        :param ex: The Exception to be filled
+        """
+        import traceback
+
+        self.proto.code = jina_pb2.StatusProto.ERROR
+        self.proto.description = repr(ex)
+        self.proto.exception.name = ex.__class__.__name__
+        self.proto.exception.args.extend([str(v) for v in ex.args])
+        self.proto.exception.stacks.extend(
+            traceback.format_exception(etype=type(ex), value=ex, tb=ex.__traceback__)
+        )
+
+    def set_code(self, code):
+        """Set the code of the Status Message
+
+        :param code: The code to be added
+        """
+        self.proto.code = code
diff --git a/tests/integration/flow-dry-run/__init__.py b/tests/integration/flow-dry-run/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/flow-dry-run/test_flow_dry_run.py b/tests/integration/flow-dry-run/test_flow_dry_run.py
new file mode 100644
index 0000000000000..851cf5f4d0f73
--- /dev/null
+++ b/tests/integration/flow-dry-run/test_flow_dry_run.py
@@ -0,0 +1,14 @@
+import pytest
+
+from jina import Flow
+
+
+@pytest.mark.parametrize('protocol', ['grpc', 'http', 'websocket'])
+def test_dry_run(protocol):
+    f = Flow(protocol=protocol).add()
+    with f:
+        dry_run = f.dry_run()
+    dry_run_negative = f.dry_run()
+
+    assert dry_run
+    assert not dry_run_negative
diff --git a/tests/integration/runtimes/test_gateway_dry_run.py b/tests/integration/runtimes/test_gateway_dry_run.py
new file mode 100644
index 0000000000000..00f9c833dba7a
--- /dev/null
+++ b/tests/integration/runtimes/test_gateway_dry_run.py
@@ -0,0 +1,99 @@
+import asyncio
+import json
+import multiprocessing
+import threading
+import time
+from collections import defaultdict
+
+import pytest
+
+from jina import Client, Document, Executor, requests
+from jina.enums import PollingType
+from jina.parsers import set_gateway_parser, set_pod_parser
+from jina.serve.runtimes.asyncio import AsyncNewLoopRuntime
+from jina.serve.runtimes.gateway.grpc import GRPCGatewayRuntime
+from jina.serve.runtimes.gateway.http import HTTPGatewayRuntime
+from jina.serve.runtimes.gateway.websocket import WebSocketGatewayRuntime
+from jina.serve.runtimes.worker import WorkerRuntime
+
+
+def _create_worker_runtime(port, name='', executor=None):
+    args = set_pod_parser().parse_args([])
+    args.port = port
+    args.name = name
+    if executor:
+        args.uses = executor
+    with WorkerRuntime(args) as runtime:
+        runtime.run_forever()
+
+
+def _create_gateway_runtime(graph_description, pod_addresses, port, protocol='grpc'):
+    if protocol == 'http':
+        gateway_runtime = HTTPGatewayRuntime
+    elif protocol == 'websocket':
+        gateway_runtime = WebSocketGatewayRuntime
+    else:
+        gateway_runtime = GRPCGatewayRuntime
+    with gateway_runtime(
+        set_gateway_parser().parse_args(
+            [
+                '--graph-description',
+                graph_description,
+                '--deployments-addresses',
+                pod_addresses,
+                '--port',
+                str(port),
+            ]
+        )
+    ) as runtime:
+        runtime.run_forever()
+
+
+@pytest.mark.parametrize('protocol', ['grpc', 'http', 'websocket'])
+def test_dry_run_of_flow(port_generator, protocol):
+    worker_port = port_generator()
+    port = port_generator()
+    graph_description = '{"start-gateway": ["pod0"], "pod0": ["end-gateway"]}'
+    pod_addresses = f'{{"pod0": ["0.0.0.0:{worker_port}"]}}'
+
+    # create a single worker runtime
+    worker_process = multiprocessing.Process(
+        target=_create_worker_runtime, args=(worker_port,)
+    )
+    worker_process.start()
+
+    # create a single gateway runtime
+    gateway_process = multiprocessing.Process(
+        target=_create_gateway_runtime,
+        args=(graph_description, pod_addresses, port, protocol),
+    )
+    gateway_process.start()
+
+    AsyncNewLoopRuntime.wait_for_ready_or_shutdown(
+        timeout=5.0,
+        ctrl_address=f'0.0.0.0:{worker_port}',
+        ready_or_shutdown_event=multiprocessing.Event(),
+    )
+    AsyncNewLoopRuntime.wait_for_ready_or_shutdown(
+        timeout=5.0,
+        ctrl_address=f'0.0.0.0:{port}',
+        ready_or_shutdown_event=multiprocessing.Event(),
+    )
+
+    # send requests to the gateway
+    c = Client(host='localhost', port=port, asyncio=True, protocol=protocol)
+    dry_run_alive = c.dry_run()
+
+    worker_process.terminate()
+    worker_process.join()
+
+    dry_run_worker_removed = c.dry_run()
+
+    gateway_process.terminate()
+    gateway_process.join()
+
+    assert dry_run_alive
+    assert not dry_run_worker_removed
+
+    assert gateway_process.exitcode == 0
+    assert worker_process.exitcode == 0
diff --git a/tests/unit/serve/executors/test_executor.py b/tests/unit/serve/executors/test_executor.py
index 98145dc942eba..eb1e869c5e222 100644
--- a/tests/unit/serve/executors/test_executor.py
+++ b/tests/unit/serve/executors/test_executor.py
@@ -254,6 +254,10 @@ def do(self, *args, **kwargs):
     ],
 )
 def test_override_requests(uses_requests, expected):
+    from jina.serve.executors import __dry_run_endpoint__
+
+    expected.add(__dry_run_endpoint__)
+
     class OverrideExec(Executor):
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
diff --git a/tests/unit/types/request/test_status_message.py b/tests/unit/types/request/test_status_message.py
new file mode 100644
index 0000000000000..46bfa63833a4d
--- /dev/null
+++ b/tests/unit/types/request/test_status_message.py
@@ -0,0 +1,41 @@
+import pytest
+from google.protobuf.json_format import MessageToDict, MessageToJson
+
+from jina.excepts import BadRequestType
+from jina.proto import jina_pb2
+from jina.types.request.status import StatusMessage
+
+
+@pytest.fixture(scope='function')
+def status_pb():
+    return jina_pb2.StatusProto()
+
+
+def test_init(status_pb):
+    assert StatusMessage(status_object=None)
+    assert StatusMessage(status_object=status_pb)
+    assert StatusMessage(status_object=MessageToDict(status_pb))
+    assert StatusMessage(status_object=MessageToJson(status_pb))
+
+
+def test_init_fail():
+    with pytest.raises(BadRequestType):
+        StatusMessage(status_object=5)
+
+
+@pytest.mark.parametrize(
+    'status_code', [jina_pb2.StatusProto.SUCCESS, jina_pb2.StatusProto.ERROR]
+)
+def test_set_code(status_code):
+    status = StatusMessage()
+    status.set_code(status_code)
+    assert status.proto.code == status_code
+
+
+def test_set_exception():
+    status = StatusMessage()
+    exc = Exception('exception code')
+    status.set_exception(exc)
+    assert status.proto.code == jina_pb2.StatusProto.ERROR
+    assert status.proto.description == repr(exc)
+    assert status.proto.exception.name == exc.__class__.__name__


Code review comments:
diff_hunk: @@ -0,0 +1,238 @@
+(health-check-executor)=
+# Health check of an Executor
+
+Executors run as microservices exposing `grpc` endpoints. To give information to the outside world about their health and their readiness to receive requests,
+Executors expose a [grpc health check](https://github.com/grpc/grpc/blob/master/doc/health-checking.md) service which can be used by different orchestrators like docker-compose, 
+or by other services like load-balancers. Jina itself when run locally uses this service to make sure that each Executor is ready to receive traffic.
+
+To see how to use it, let's start a Flow inside a Terminal and let it blocked to serve requests:
+
+```python
+from jina import Flow
+
+f = Flow(protocol='grpc', port=12345).add(port=12346)
+with f:
+    f.block()
+```
+
+On another terminal, we can install or download [grpcurl](https://github.com/fullstorydev/grpcurl) image to be able to send `rpc` requests to our services.
+
+```bash
+docker pull fullstorydev/grpcurl:latest
+docker run --network='host' fullstorydev/grpcurl -plaintext 127.0.0.1:12346 grpc.health.v1.Health/Check
+```
+
+```text
+{
+  "status": "SERVING"
+}
+```
+
+(health-check-gateway)=
+# Health check of the Gateway
+
+The same way individual Executors expose endpoints for orchestrators, clients or other services to check their availability, Gateway as a microservice also exposes this in different ways and depending
+on the protocol used.
+
+
+## Gateway health check with grpc
+
+When using grpc as the protocol used to communicate with the Gateway, then Gateway uses the exact same mechanism as Executors to expose their individual health status. It exposes [grpc health check](https://github.com/grpc/grpc/blob/master/doc/health-checking.md) to the outside world.
+
+With the same Flow described before we can use the same way to check the gateway status:
+
+```bash
+docker run --network='host' fullstorydev/grpcurl -plaintext 127.0.0.1:12345 grpc.health.v1.Health/Check
+```
+
+```text
+{
+  "status": "SERVING"
+}
+```
+
+
+## Gateway health check with http or websocket
+
+When using grpc or http as a procotol for the Gateway, then it exposes and endpoint '/' that one can query to check the status.
+
+By having a Flow with http or websocket as protocol:
+
+```python
+from jina import Flow
+
+f = Flow(protocol='http', port=12345).add()
+with f:
+    f.block()
+```
+
+```bash
+curl http://localhost:12345
+```
+
+Then you will get a valid empty response indicating its ability to serve.
+```text
+{}%
+```
+
+# Readiness of a Flow exposed to the client
+
+A lot of times, from the client perspective, it is useful to know if a Flow as a complete set of microservices is ready to receive requests. This is why the gateway 
+exposes an endpoint for each of the supported protocols to know the health and readiness of a Flow. 
+
+Also Jina Flow and Client contains a convenient API to query these endpoints. You can call `flow.dry_run()` and `client.dry_run()`
+
+````{tab} via Flow
+```python
+from jina import Flow
+
+with Flow().add() as f:
+    print(f.dry_run())
+
+print(f.dry_run())
+```
+```text
+True
+False
+```
+````
+````{tab} via Client
+```python
+from jina import Flow
+
+with Flow(port=12345).add() as f:
+    f.block()
+```
+```python
+from jina import Client
+
+client = Client(port=12345)
+print(client.dry_run())
+```
+```text
+True
+```
+````
+
+## Check status from outside
+
+To understand better what is going on under the hood, we can check how to target the endpoint from outside Jina or the client.
+
+Let's first instantiate the Flow with its corresponding protocol and block it for serving.
+
+```python
+from jina import Flow
+import os
+
+PROTOCOL = 'grpc'  # it could also be http or websocket
+
+os.setenv[
+    'JINA_LOG_LEVEL'
+] = 'DEBUG'  # this way we can check what is the PID of the Executor
+
+with Flow(protocol=PROTOCOL, port=12345).add() as f:
+    f.block()
+```
+
+```text
+⠋  Waiting ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0/0 -:--:--DEBUG  gateway/rep-0@19075 adding connection for deployment executor0/heads/0 to grpc://0.0.0.0:12346                                                                                           [05/31/22 18:10:16]
+DEBUG  executor0/rep-0@19074 start listening on 0.0.0.0:12346                                                                                                                                   [05/31/22 18:10:16]
+DEBUG  gateway/rep-0@19075 start server bound to 0.0.0.0:12345                                                                                                                                  [05/31/22 18:10:17]
+DEBUG  executor0/rep-0@19059 ready and listening                                                                                                                                                [05/31/22 18:10:17]
+DEBUG  gateway/rep-0@19059 ready and listening                                                                                                                                                  [05/31/22 18:10:17]
+╭────── 🎉 Flow is ready to serve! ──────╮
+│  🔗  Protocol                  GRPC    │
+│  🏠     Local         0.0.0.0:12345    │
+│  🔒   Private    192.168.1.13:12345    │
+│  🌍    Public  92.178.210.177:12345    │
path: docs/fundamentals/health-check.md
text: @user1:
```suggestion
```

Maybe you don't want you personnal IP to be on the documentation