Repository: jina-ai/serve
Pull request: feat: distributed replicas across different hosts
## Goal

Allow replicas of the same Executor to run in a distributed manner (on different machines).
Closes #5103
Closes #5231 

## Usage

```python
f.add(host='localhost,91.198.174.192', port='12345,12346', external=True)
```
Alternatively:
```python
f.add(host='localhost:12345,91.198.174.192:12346', external=True)
```
will connect to `grpc://localhost:12345` and `grpc://91.198.174.192:12346` as two replicas of the same Executor.

## ToDo
- [x] allow different hosts for replicas of the same Executor
- [x] expose this option to the user
- [x] make it work in combination with passing the entire address to `host`
- [ ] what about shards?
- [x] testing
  - [x] unit tests for new helpers
  - [x] simple case
    - [x] passing `host` an `port` separately
    - [x] parsing whole schema from `host`~~ separate bug
  - [x] using docker
- [x] Documentation
  - [x] User guide, including distinction between external and shared executor
  - [x] docstring

Problem statement:
How can I serve replicas of an executor in different hosts? 
First, I thank Jina for an easy deployment/scaling framework. In my scenario, I have independent physical instance GPU resources, so is there any way to host replicas of an executor type in a different place? For example, is a `external replica` exist? Or suppose I have to do that manually. In that case, any tool in the Jina ecosystem or third party can help me to solve this problem?

Bug: For external Executors, passing entire address to `host` does not work
The following syntax, which is supported for the gateway, does not work for external executors:

```python
f.add(host='grpc://localhost:1234', external=True)
```

Instead, host and port have to be precised separately:

```python
f.add(host='localhost', port=1234, external=True)
```

Not that this bug also applies to replicated external Executors:

```python
f.add(host='grpc://localhost:1234,grpc://localhost:1235', external=True)
```

Hints from issue discussion:
oh, @JoanFM @samsja @JohannesMessner remember what I told you 2 months ago in the office? The load balancer idea 😉 

let's do it,let's do it,let's do it, i feel excited 😆 
Hey @nhtlongcs,

This is not possible right now. This solution for now is delivered via Kubernetes where K8s would schedule different Pods in different machines. 
@hanxiao @JoanFM Thanks for your replies, hope this feature will available soon. I have learned many about common ml-ops practices and system design from Jina
let's keep the ticket open, i will push this feature to be landed 
Hey @nhtlongcs, my question is how u plan to orchestrate these different Replicas in different machines? If one of these replicas go down, do you need it to be automatically restarted, etc...? This is why we rely on K8s to be our most professional and advanced Container Orchestrator rather than reinventing the wheel ourselves. 


Hi @JoanFM. In my situation, when opening this issue, I only expected Jina could help me with horizontal scaling easiest and simplest way. For example, in my case, if one of my replicas goes down, the load balancer can redirect automatically from 4 to 3 (for example) and notify me to take some manual actions (or may I can monitor that), which is acceptable (in my opinion). 

Honestly, I have never done this before. A professional tool designed to solve multiple scenarios, such as k8s, is too advanced and challenging for me to utilize in my specific case (but thank you for mentioning that, I will find out how to use it). Maybe that is one of the reasons that led me to choose Jina instead of many professional DevOps platforms to achieve my scope.
Okey, we will take into consideration when designing this feature, thanks
Raised by community is added because it was raised by community and me
I have started an implementation at https://github.com/jina-ai/jina/pull/5217.
Let's move more implementation specific discussions there.

Commit message: test: add test for externial replicas in container

Patch to review:
diff --git a/docs/fundamentals/flow/executor-args.md b/docs/fundamentals/flow/executor-args.md
index 9502cf34d82d0..76e33a6a2c872 100644
--- a/docs/fundamentals/flow/executor-args.md
+++ b/docs/fundamentals/flow/executor-args.md
@@ -12,7 +12,7 @@
 | `uses_metas` | Dictionary of keyword arguments that will override the `metas` configuration in `uses` | `object` | `None` |
 | `uses_requests` | Dictionary of keyword arguments that will override the `requests` configuration in `uses` | `object` | `None` |
 | `py_modules` | The customized python modules need to be imported before loading the executor<br><br>Note that the recommended way is to only import a single module - a simple python file, if your<br>executor can be defined in a single file, or an ``__init__.py`` file if you have multiple files,<br>which should be structured as a python package. For more details, please see the<br>`Executor cookbook <https://docs.jina.ai/fundamentals/executor/executor-files/>`__ | `array` | `None` |
-| `port` | The port for input data to bind to, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
+| `port` | The port for input data to bind to, default is a random port between [49152, 65535] | `string` | `random in [49152, 65535]` |
 | `host_in` | The host address for binding to, by default it is 0.0.0.0 | `string` | `0.0.0.0` |
 | `native` | If set, only native Executors is allowed, and the Executor is always run inside WorkerRuntime. | `boolean` | `False` |
 | `output_array_type` | The type of array `tensor` and `embedding` will be serialized to.<br><br>Supports the same types as `docarray.to_protobuf(.., ndarray_type=...)`, which can be found <br>`here <https://docarray.jina.ai/fundamentals/document/serialization/#from-to-protobuf>`.<br>Defaults to retaining whatever type is returned by the Executor. | `string` | `None` |
diff --git a/docs/fundamentals/flow/gateway-args.md b/docs/fundamentals/flow/gateway-args.md
index ea33e2a32a00c..e287e4f7d084b 100644
--- a/docs/fundamentals/flow/gateway-args.md
+++ b/docs/fundamentals/flow/gateway-args.md
@@ -12,7 +12,7 @@
 | `uses_metas` | Dictionary of keyword arguments that will override the `metas` configuration in `uses` | `object` | `None` |
 | `uses_requests` | Dictionary of keyword arguments that will override the `requests` configuration in `uses` | `object` | `None` |
 | `py_modules` | The customized python modules need to be imported before loading the executor<br><br>Note that the recommended way is to only import a single module - a simple python file, if your<br>executor can be defined in a single file, or an ``__init__.py`` file if you have multiple files,<br>which should be structured as a python package. For more details, please see the<br>`Executor cookbook <https://docs.jina.ai/fundamentals/executor/executor-files/>`__ | `array` | `None` |
-| `port` | The port for input data to bind to, default is a random port between [49152, 65535] | `number` | `random in [49152, 65535]` |
+| `port` | The port for input data to bind to, default is a random port between [49152, 65535] | `string` | `random in [49152, 65535]` |
 | `host_in` | The host address for binding to, by default it is 0.0.0.0 | `string` | `0.0.0.0` |
 | `native` | If set, only native Executors is allowed, and the Executor is always run inside WorkerRuntime. | `boolean` | `False` |
 | `output_array_type` | The type of array `tensor` and `embedding` will be serialized to.<br><br>Supports the same types as `docarray.to_protobuf(.., ndarray_type=...)`, which can be found <br>`here <https://docarray.jina.ai/fundamentals/document/serialization/#from-to-protobuf>`.<br>Defaults to retaining whatever type is returned by the Executor. | `string` | `None` |
diff --git a/jina/helper.py b/jina/helper.py
index 28341d08d1c8d..e7723f2e07e37 100644
--- a/jina/helper.py
+++ b/jina/helper.py
@@ -966,7 +966,8 @@ def get_full_version() -> Optional[Tuple[Dict, Dict]]:
             'session-id': str(random_uuid(use_uuid1=True)),
             'uptime': __uptime__,
             'ci-vendor': get_ci_vendor() or __unset_msg__,
-            'internal': 'jina-ai' in os.getenv('GITHUB_ACTION_REPOSITORY', __unset_msg__)
+            'internal': 'jina-ai'
+            in os.getenv('GITHUB_ACTION_REPOSITORY', __unset_msg__),
         }
 
         env_info = {k: os.getenv(k, __unset_msg__) for k in __jina_env__}
@@ -1619,7 +1620,11 @@ def _parse_url(host):
     return scheme, host, port
 
 
-def is_port_free(host, port):
+def is_port_free(host: str, port: Union[int, str]) -> bool:
+    try:
+        port = int(port)
+    except ValueError:
+        raise ValueError(f'port {port} is not an integer and cannot be cast to one')
     with socket(AF_INET, SOCK_STREAM) as session:
         if session.connect_ex((host, port)) == 0:
             return False
@@ -1638,7 +1643,7 @@ def _parse_ports(port: str) -> Union[int, List]:
         _parse_port('8000')
         8000
 
-        _parse_port('8001:8002:8005')
+        _parse_port('8001,8002,8005')
         [80001, 8002, 8005]
 
     :param port: the string to parse
@@ -1654,6 +1659,27 @@ def _parse_ports(port: str) -> Union[int, List]:
     return port
 
 
+def _parse_hosts(host: str) -> Union[str, List[str]]:
+    """Parse port
+
+    EXAMPLE USAGE
+
+    .. code-block:: python
+
+
+        _parse_hosts('localhost')
+        'localhost'
+
+        _parse_port('localhost,91.198.174.192')
+        ['localhost', '91.198.174.192']
+
+    :param host: the string to parse
+    :return: the host or the iterable of hosts
+    """
+    hosts = host.split(',')
+    return hosts[0] if len(hosts) == 1 else hosts
+
+
 def send_telemetry_event(event: str, obj: Any, **kwargs) -> None:
     """Sends in a thread a request with telemetry for a given event
     :param event: Event leading to the telemetry entry
@@ -1685,3 +1711,36 @@ def _telemetry():
             pass
 
     threading.Thread(target=_telemetry, daemon=True).start()
+
+
+def make_iterable(o: object) -> Iterable:
+    """
+    Make an object an iterable by wrapping it as a singleton list.
+    If the input is already an iterable (except str and bytes), it will be returned as is.
+    Str and bytes are treated as non-iterable, and thus wrapped in a list.
+
+    EXAMPLE USAGE
+
+    .. code-block:: python
+
+
+        make_iter(1)
+        [1]
+
+        make_iter('a')
+        ['a']
+
+        make_iter([1, 2, 3])
+        [1, 2, 3]
+
+        make_iter((1, 2, 3))
+        (1, 2, 3)
+
+
+    :param o: the object to be converted to an iterable
+    :return: the iterable
+    """
+    if isinstance(o, Iterable) and not isinstance(o, (str, bytes)):
+        return o
+    else:
+        return [o]
diff --git a/jina/orchestrate/deployments/__init__.py b/jina/orchestrate/deployments/__init__.py
index 2fb40d5a20027..2c92020d3511c 100644
--- a/jina/orchestrate/deployments/__init__.py
+++ b/jina/orchestrate/deployments/__init__.py
@@ -12,7 +12,12 @@
 
 from jina import __default_executor__, __default_host__, __docker_host__, helper
 from jina.enums import DeploymentRoleType, PodRoleType, PollingType
-from jina.helper import CatchAllCleanupContextManager, _parse_ports
+from jina.helper import (
+    CatchAllCleanupContextManager,
+    _parse_hosts,
+    _parse_ports,
+    make_iterable,
+)
 from jina.hubble.helper import replace_secret_of_hub_uri
 from jina.hubble.hubio import HubIO
 from jina.jaml.helper import complete_path
@@ -251,6 +256,12 @@ def __init__(
             needs or set()
         )  #: used in the :class:`jina.flow.Flow` to build the graph
 
+        (
+            self.ext_repl_hosts,
+            self.ext_repl_ports,
+        ) = self._parse_external_replica_hosts_and_ports()
+        if len(self.ext_repl_ports) > 1:
+            self.args.replicas = len(self.ext_repl_ports)
         self.uses_before_pod = None
         self.uses_after_pod = None
         self.head_pod = None
@@ -263,6 +274,19 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         super().__exit__(exc_type, exc_val, exc_tb)
         self.join()
 
+    def _parse_external_replica_hosts_and_ports(self):
+        ext_repl_ports = make_iterable(_parse_ports(str(self.args.port)))
+        ext_repl_hosts = make_iterable(_parse_hosts(str(self.args.host)))
+        if len(ext_repl_ports) != len(ext_repl_hosts):
+            if (
+                len(ext_repl_hosts) == 1
+            ):  # only one host given, assume replicas are on the same host
+                ext_repl_hosts = ext_repl_hosts * len(ext_repl_ports)
+            else:
+                raise ValueError('Number of ports does not match number of hosts')
+        self.args.port, self.args.host = int(ext_repl_ports[0]), ext_repl_hosts[0]
+        return ext_repl_hosts, ext_repl_ports
+
     def _update_port_args(self):
         _all_port_monitoring = _parse_ports(self.args.port_monitoring)
         self.args.all_port_monitoring = (
@@ -282,6 +306,13 @@ def update_pod_args(self):
         else:
             self.pod_args = self._parse_args(self.args)
 
+        if self.external:
+            for pod, port, host in zip(
+                self.pod_args['pods'][0], self.ext_repl_ports, self.ext_repl_hosts
+            ):
+                pod.port = port
+                pod.host = host
+
     def update_sandbox_args(self):
         """Update args of all its pods based on the host and port returned by Hubble"""
         if self.is_sandbox:
@@ -390,6 +421,19 @@ def ports(self) -> List[int]:
                 ports.append(replica.port)
             return ports
 
+    @property
+    def hosts(self) -> List[str]:
+        """Returns a list of host addresses exposed by this Deployment.
+        Exposed means these are the host a Client/Gateway is supposed to communicate with.
+        For sharded deployments this will be the head host.
+        For non sharded deployments it will be all replica hosts
+        .. # noqa: DAR201
+        """
+        if self.head_host:
+            return [self.head_host]
+        else:
+            return [replica.host for replica in self.pod_args['pods'][0]]
+
     @property
     def dockerized_uses(self) -> bool:
         """Checks if this Deployment uses a dockerized Executor
@@ -674,7 +718,9 @@ def _roundrobin_cuda_device(device_str: str, replicas: int):
 
             selected_devices = []
             if device_str[2:]:
-                for device_num in Deployment._parse_devices(device_str[2:], num_devices):
+                for device_num in Deployment._parse_devices(
+                    device_str[2:], num_devices
+                ):
                     selected_devices.append(device_num)
             else:
                 selected_devices = range(num_devices)
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 4181b7ea82ab4..8acf6ee786c56 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -54,10 +54,13 @@
 from jina.helper import (
     ArgNamespace,
     CatchAllCleanupContextManager,
+    _parse_hosts,
+    _parse_ports,
     download_mermaid_url,
     get_internal_ip,
     get_public_ip,
     is_port_free,
+    make_iterable,
     send_telemetry_event,
     typename,
 )
@@ -165,7 +168,7 @@ def __init__(
         no_debug_endpoints: Optional[bool] = False,
         output_array_type: Optional[str] = None,
         polling: Optional[str] = 'ANY',
-        port: Optional[int] = None,
+        port: Optional[str] = None,
         port_monitoring: Optional[str] = None,
         prefetch: Optional[int] = 1000,
         protocol: Optional[str] = 'GRPC',
@@ -621,18 +624,29 @@ def _add_gateway(
 
     def _get_deployments_addresses(self) -> Dict[str, List[str]]:
         graph_dict = {}
-        for node, v in self._deployment_nodes.items():
+        for node, deployment in self._deployment_nodes.items():
             if node == 'gateway':
                 continue
-            if v.head_args:
+            if deployment.head_args:
                 # add head information
-                graph_dict[node] = [f'{v.protocol}://{v.host}:{v.head_port}']
+                graph_dict[node] = [
+                    f'{deployment.protocol}://{deployment.host}:{deployment.head_port}'
+                ]
             else:
                 # there is no head, add the worker connection information instead
-                host = v.host
-                if host_is_local(host) and in_docker() and v.dockerized_uses:
-                    host = __docker_host__
-                graph_dict[node] = [f'{v.protocol}://{host}:{port}' for port in v.ports]
+                ports = deployment.ports
+                hosts = [
+                    __docker_host__
+                    if host_is_local(host)
+                    and in_docker()
+                    and deployment.dockerized_uses
+                    else host
+                    for host in deployment.hosts
+                ]
+                graph_dict[node] = [
+                    f'{deployment.protocol}://{host}:{port}'
+                    for host, port in zip(hosts, ports)
+                ]
 
         return graph_dict
 
@@ -809,7 +823,7 @@ def add(
         native: Optional[bool] = False,
         output_array_type: Optional[str] = None,
         polling: Optional[str] = 'ANY',
-        port: Optional[int] = None,
+        port: Optional[str] = None,
         port_monitoring: Optional[str] = None,
         py_modules: Optional[List[str]] = None,
         quiet: Optional[bool] = False,
@@ -1181,6 +1195,20 @@ def add(
                 'External Executors with multiple needs have to do auto reduce.'
             )
 
+        # args.external_replica_ports = make_iterable(_parse_ports(str(args.port)))
+        # args.external_replica_hosts = make_iterable(_parse_hosts(str(args.host)))
+
+        # if len(args.external_replica_ports) != len(args.external_replica_hosts):
+        #     if (
+        #         len(args.external_replica_hosts) == 1
+        #     ):  # only one host given, assume replicas are on the same host
+        #         args.external_replica_hosts = args.external_replica_hosts * len(
+        #             args.external_replica_ports
+        #         )
+        #     else:
+        #         raise ValueError('Number of ports does not match number of hosts')
+        # if len(args.external_replica_ports) > 1:
+        #     args.replicas = len(args.external_replica_ports)
         op_flow._deployment_nodes[deployment_name] = Deployment(args, needs)
 
         if not args.floating:
@@ -1479,9 +1507,9 @@ def start(self):
             for k, v in self.args.env.items():
                 os.environ[k] = str(v)
 
-        for k, v in self:
-            if not v.external:
-                self.enter_context(v)
+        for depl_name, deployment in self:
+            if not deployment.external:
+                self.enter_context(deployment)
 
         self._wait_until_all_ready()
 
diff --git a/jina/parsers/orchestrate/pod.py b/jina/parsers/orchestrate/pod.py
index 612a15e32fa2a..3dbd232f93460 100644
--- a/jina/parsers/orchestrate/pod.py
+++ b/jina/parsers/orchestrate/pod.py
@@ -84,8 +84,8 @@ def mixin_pod_parser(parser):
 
     gp.add_argument(
         '--port',
-        type=int,
-        default=helper.random_port(),
+        type=str,
+        default=str(helper.random_port()),
         help='The port for input data to bind to, default is a random port between [49152, 65535]',
     )
 
diff --git a/jina/serve/runtimes/base.py b/jina/serve/runtimes/base.py
index 1717b63bac705..e75a740c56135 100644
--- a/jina/serve/runtimes/base.py
+++ b/jina/serve/runtimes/base.py
@@ -56,6 +56,7 @@ def __init__(
     ):
         super().__init__()
         self.args = args
+        self.args.port = int(self.args.port)
         if args.name:
             self.name = f'{args.name}/{self.__class__.__name__}'
         else:
diff --git a/tests/integration/external_deployment/Dockerfile b/tests/integration/external_deployment/Dockerfile
new file mode 100644
index 0000000000000..432d2f113aeef
--- /dev/null
+++ b/tests/integration/external_deployment/Dockerfile
@@ -0,0 +1,7 @@
+FROM jinaai/jina:master
+
+COPY . /executor_root/
+
+WORKDIR /executor_root
+
+ENTRYPOINT ["jina", "executor", "--uses", "config.yml"]
\ No newline at end of file
diff --git a/tests/integration/external_deployment/config.yml b/tests/integration/external_deployment/config.yml
new file mode 100644
index 0000000000000..d153db80edd7e
--- /dev/null
+++ b/tests/integration/external_deployment/config.yml
@@ -0,0 +1,3 @@
+jtype: MyExternalExecutor
+py_modules:
+ - exec.py
\ No newline at end of file
diff --git a/tests/integration/external_deployment/exec.py b/tests/integration/external_deployment/exec.py
new file mode 100644
index 0000000000000..9c2ce308f9532
--- /dev/null
+++ b/tests/integration/external_deployment/exec.py
@@ -0,0 +1,15 @@
+import uuid
+
+from jina import Executor, requests
+
+
+class MyExternalExecutor(Executor):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._id = str(uuid.uuid4())
+
+    @requests
+    def foo(self, docs, *args, **kwargs):
+        for doc in docs:
+            doc.tags['name'] = self.runtime_args.name
+            doc.tags['uuid'] = self._id
diff --git a/tests/integration/external_deployment/test_external_deployment.py b/tests/integration/external_deployment/test_external_deployment.py
index 5c02498e4f01d..b2981c0a733d5 100644
--- a/tests/integration/external_deployment/test_external_deployment.py
+++ b/tests/integration/external_deployment/test_external_deployment.py
@@ -1,3 +1,7 @@
+import os
+import time
+import uuid
+
 import pytest
 
 from jina import Document, DocumentArray, Executor, Flow, requests
@@ -5,6 +9,8 @@
 from jina.orchestrate.deployments import Deployment
 from jina.parsers import set_deployment_parser
 
+cur_dir = os.path.dirname(os.path.abspath(__file__))
+
 
 def validate_response(docs, expected_docs=50):
     assert len(docs) == expected_docs
@@ -22,15 +28,14 @@ def num_shards(request):
     return request.param
 
 
-@pytest.fixture(scope='function')
-def external_deployment_args(num_shards):
+def _external_deployment_args(num_shards, port=None):
     args = [
         '--uses',
         'MyExternalExecutor',
         '--name',
         'external_real',
         '--port',
-        str(random_port()),
+        str(port) if port else str(random_port()),
         '--host-in',
         '0.0.0.0',
         '--shards',
@@ -41,6 +46,42 @@ def external_deployment_args(num_shards):
     return set_deployment_parser().parse_args(args)
 
 
+def _external_deployment_args_docker(num_shards, port=None):
+    args = [
+        '--uses',
+        'docker://worker-runtime',
+        '--name',
+        'external_real',
+        '--port',
+        str(port) if port else str(random_port()),
+        '--host-in',
+        '0.0.0.0',
+        '--shards',
+        str(num_shards),
+        '--polling',
+        'all',
+    ]
+    return set_deployment_parser().parse_args(args)
+
+
+@pytest.fixture
+def replica_docker_image_built():
+    import docker
+
+    client = docker.from_env()
+    client.images.build(path=cur_dir, tag='worker-runtime')
+    client.close()
+    yield
+    time.sleep(2)
+    client = docker.from_env()
+    client.containers.prune()
+
+
+@pytest.fixture(scope='function')
+def external_deployment_args(num_shards, port=None):
+    return _external_deployment_args(num_shards, port)
+
+
 @pytest.fixture
 def external_deployment(external_deployment_args):
     return Deployment(external_deployment_args)
@@ -49,16 +90,18 @@ def external_deployment(external_deployment_args):
 class MyExternalExecutor(Executor):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        self._id = str(uuid.uuid4())
 
     @requests
     def foo(self, docs, *args, **kwargs):
         for doc in docs:
             doc.tags['name'] = self.runtime_args.name
+            doc.tags['uuid'] = self._id
 
 
 @pytest.mark.parametrize('num_shards', [1, 2], indirect=True)
 def test_flow_with_external_deployment(
-        external_deployment, external_deployment_args, input_docs, num_shards
+    external_deployment, external_deployment_args, input_docs, num_shards
 ):
     with external_deployment:
         external_args = vars(external_deployment_args)
@@ -80,7 +123,7 @@ def test_flow_with_external_deployment(
 
 @pytest.mark.parametrize('num_shards', [2], indirect=True)
 def test_two_flow_with_shared_external_deployment(
-        external_deployment, external_deployment_args, input_docs, num_shards
+    external_deployment, external_deployment_args, input_docs, num_shards
 ):
     external_deployment.head_args.disable_reduce = True
     with external_deployment:
@@ -96,8 +139,8 @@ def test_two_flow_with_shared_external_deployment(
 
         flow2 = (
             Flow()
-                .add(name='foo')
-                .add(
+            .add(name='foo')
+            .add(
                 **external_args,
                 name='external_fake',
                 external=True,
@@ -161,12 +204,12 @@ def external_deployment_shards_2(external_deployment_shards_2_args):
 
 @pytest.mark.parametrize('num_shards', [1, 2], indirect=True)
 def test_flow_with_external_deployment_shards(
-        external_deployment_shards_1,
-        external_deployment_shards_2,
-        external_deployment_shards_1_args,
-        external_deployment_shards_2_args,
-        input_docs,
-        num_shards,
+    external_deployment_shards_1,
+    external_deployment_shards_2,
+    external_deployment_shards_1_args,
+    external_deployment_shards_2_args,
+    input_docs,
+    num_shards,
 ):
     with external_deployment_shards_1, external_deployment_shards_2:
         external_args_1 = vars(external_deployment_shards_1_args)
@@ -179,20 +222,20 @@ def test_flow_with_external_deployment_shards(
         del external_args_2['deployment_role']
         flow = (
             Flow()
-                .add(name='executor1')
-                .add(
+            .add(name='executor1')
+            .add(
                 **external_args_1,
                 name='external_fake_1',
                 external=True,
                 needs=['executor1'],
             )
-                .add(
+            .add(
                 **external_args_2,
                 name='external_fake_2',
                 external=True,
                 needs=['executor1'],
             )
-                .needs(needs=['external_fake_1', 'external_fake_2'], port=random_port())
+            .needs(needs=['external_fake_1', 'external_fake_2'], port=random_port())
         )
 
         with flow:
@@ -226,10 +269,10 @@ def external_deployment_pre_shards(external_deployment_pre_shards_args):
 
 @pytest.mark.parametrize('num_shards', [1, 2], indirect=True)
 def test_flow_with_external_deployment_pre_shards(
-        external_deployment_pre_shards,
-        external_deployment_pre_shards_args,
-        input_docs,
-        num_shards,
+    external_deployment_pre_shards,
+    external_deployment_pre_shards_args,
+    input_docs,
+    num_shards,
 ):
     with external_deployment_pre_shards:
         external_args = vars(external_deployment_pre_shards_args)
@@ -238,20 +281,20 @@ def test_flow_with_external_deployment_pre_shards(
         del external_args['deployment_role']
         flow = (
             Flow()
-                .add(
+            .add(
                 **external_args,
                 name='external_fake',
                 external=True,
             )
-                .add(
+            .add(
                 name='executor1',
                 needs=['external_fake'],
             )
-                .add(
+            .add(
                 name='executor2',
                 needs=['external_fake'],
             )
-                .needs(['executor1', 'executor2'])
+            .needs(['executor1', 'executor2'])
         )
         with flow:
             resp = flow.index(inputs=input_docs)
@@ -286,10 +329,10 @@ def external_deployment_join(external_deployment_join_args):
 
 @pytest.mark.parametrize('num_shards', [1, 2], indirect=True)
 def test_flow_with_external_deployment_join(
-        external_deployment_join,
-        external_deployment_join_args,
-        input_docs,
-        num_shards,
+    external_deployment_join,
+    external_deployment_join_args,
+    input_docs,
+    num_shards,
 ):
     with external_deployment_join:
         external_args = vars(external_deployment_join_args)
@@ -298,19 +341,19 @@ def test_flow_with_external_deployment_join(
         del external_args['deployment_role']
         flow = (
             Flow()
-                .add(
+            .add(
                 **external_args,
                 external=True,
             )
-                .add(
+            .add(
                 name='executor1',
                 needs=['executor0'],
             )
-                .add(
+            .add(
                 name='executor2',
                 needs=['executor0'],
             )
-                .needs(
+            .needs(
                 **external_args,
                 external=True,
                 needs=['executor1', 'executor2'],
@@ -325,7 +368,6 @@ def test_flow_with_external_deployment_join(
 
 def test_external_flow_with_target_executor():
     class ExtExecutor(Executor):
-
         @requests
         def foo(self, docs, **kwargs):
             for doc in docs:
@@ -339,5 +381,46 @@ def foo(self, docs, **kwargs):
         with f:
             response = f.post(on='/', inputs=d, target_executor='external_executor')
 
-
     assert response[0].text == 'external'
+
+
+@pytest.mark.parametrize('hosts', ['localhost,localhost', 'localhost'])
+def test_distributed_replicas(input_docs, hosts):
+    port1, port2 = random_port(), random_port()
+    args1, args2 = _external_deployment_args(
+        num_shards=1, port=port1
+    ), _external_deployment_args(num_shards=1, port=port2)
+    depl1 = Deployment(args1)
+    depl2 = Deployment(args2)
+    with depl1, depl2:
+        flow = Flow().add(
+            host=hosts,
+            port=f'{port1},{port2}',
+            external=True,
+        )
+        with flow:
+            resp = flow.index(inputs=input_docs, request_size=2)
+
+        depl1_id = resp[0].tags['uuid']
+        assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
+
+
+@pytest.mark.parametrize('hosts', ['localhost,localhost', 'localhost'])
+def test_distributed_replicas_docker(input_docs, hosts, replica_docker_image_built):
+    port1, port2 = random_port(), random_port()
+    args1, args2 = _external_deployment_args_docker(
+        num_shards=1, port=port1
+    ), _external_deployment_args_docker(num_shards=1, port=port2)
+    depl1 = Deployment(args1)
+    depl2 = Deployment(args2)
+    with depl1, depl2:
+        flow = Flow().add(
+            host=hosts,
+            port=f'{port1},{port2}',
+            external=True,
+        )
+        with flow:
+            resp = flow.index(inputs=input_docs, request_size=2)
+
+        depl1_id = resp[0].tags['uuid']
+        assert any([depl1_id != depl_id for depl_id in resp[1:, 'tags__uuid']])
diff --git a/tests/k8s/test_k8s.py b/tests/k8s/test_k8s.py
index d1b4fb5bfc1f9..ed30185ee656a 100644
--- a/tests/k8s/test_k8s.py
+++ b/tests/k8s/test_k8s.py
@@ -124,7 +124,7 @@ async def run_test(flow, core_client, namespace, endpoint, n_docs=10, request_si
     import portforward
 
     with portforward.forward(
-        namespace, gateway_pod_name, flow.port, flow.port, config_path
+        namespace, gateway_pod_name, int(flow.port), int(flow.port), config_path
     ):
         client_kwargs = dict(
             host='localhost',
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow.py b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
index f546bcc21e50a..01ec6b266b749 100644
--- a/tests/unit/orchestrate/flow/flow-construct/test_flow.py
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
@@ -613,5 +613,5 @@ def _validate_flow(f):
 def test_set_port_deployment(port_generator):
     port = port_generator()
     with Flow().add(uses=Executor, port=port) as f:
-        assert f._deployment_nodes['executor0'].pod_args['pods'][0][0].port == port
+        assert int(f._deployment_nodes['executor0'].pod_args['pods'][0][0].port) == port
         f.index(inputs=[])
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py b/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py
index 0045382d88d15..451f7716c3b2d 100644
--- a/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow_yaml_parser.py
@@ -3,15 +3,14 @@
 
 import numpy as np
 import pytest
+from docarray.document.generators import from_ndarray
 
-from jina import Executor
+from jina import Executor, Flow
+from jina.enums import GatewayProtocolType
 from jina.excepts import BadYAMLVersion
-from jina import Flow
 from jina.jaml import JAML
-from jina.enums import GatewayProtocolType
 from jina.jaml.parsers import get_supported_versions
 from jina.parsers.flow import set_flow_parser
-from docarray.document.generators import from_ndarray
 
 cur_dir = Path(__file__).parent
 
@@ -75,7 +74,7 @@ def test_load_modify_dump_load(tmpdir):
     # assert executor args
     assert f._deployment_nodes['custom1'].args.uses == 'jinahub://CustomExecutor1'
     assert f._deployment_nodes['custom2'].args.uses == 'CustomExecutor2'
-    assert f._deployment_nodes['custom2'].args.port == 23456
+    assert int(f._deployment_nodes['custom2'].args.port) == 23456
 
     # change args inside `with`
     f.port = 12346
@@ -95,7 +94,7 @@ def test_load_modify_dump_load(tmpdir):
     # assert args modified in code
     assert f1.port == 12346
     assert f1.protocol == GatewayProtocolType.WEBSOCKET
-    assert f1._deployment_nodes['custom2'].args.port == 23457
+    assert int(f1._deployment_nodes['custom2'].args.port) == 23457
 
 
 def test_dump_load_build(monkeypatch):
@@ -123,30 +122,30 @@ def test_dump_load_build(monkeypatch):
     # these were passed by the user
     assert f.port == f1.port
     assert f.protocol == f1.protocol
-    assert f['executor1'].args.port == f1['executor1'].args.port
+    assert int(f['executor1'].args.port) == int(f1['executor1'].args.port)
     assert f['executor2'].args.host == f1['executor2'].args.host
     # this was set during `load_config`
-    assert f['executor2'].args.port == f1['executor2'].args.port
+    assert int(f['executor2'].args.port) == int(f1['executor2'].args.port)
 
     monkeypatch.setenv('JINA_FULL_CLI', 'true')
     f2: Flow = Flow.load_config(JAML.dump(f)).build()
     # these were passed by the user
-    assert f.port == f2.port
+    assert int(f.port) == int(f2.port)
     # validate gateway args (set during build)
-    assert f['gateway'].args.port == f2['gateway'].args.port
+    assert int(f['gateway'].args.port) == int(f2['gateway'].args.port)
 
 
 def test_load_flow_with_port():
     f = Flow.load_config('yaml/test-flow-port.yml')
     with f:
-        assert f.port == 12345
+        assert int(f.port) == 12345
 
 
 def test_load_flow_from_cli():
     a = set_flow_parser().parse_args(['--uses', 'yaml/test-flow-port.yml'])
     f = Flow.load_config(a.uses)
     with f:
-        assert f.port == 12345
+        assert int(f.port) == 12345
 
 
 def test_load_flow_from_yaml():
diff --git a/tests/unit/orchestrate/flow/flow-orchestrate/test_flow_change_gateway.py b/tests/unit/orchestrate/flow/flow-orchestrate/test_flow_change_gateway.py
index dbb142831fe92..bf1a73e684ac7 100644
--- a/tests/unit/orchestrate/flow/flow-orchestrate/test_flow_change_gateway.py
+++ b/tests/unit/orchestrate/flow/flow-orchestrate/test_flow_change_gateway.py
@@ -23,8 +23,8 @@ def test_get_set_client_gateway_in_flow(protocol):
     f = Flow(protocol=protocol, port=12345)
     assert f.client_args.protocol == GatewayProtocolType.from_string(protocol)
     assert f.gateway_args.protocol == GatewayProtocolType.from_string(protocol)
-    assert f.client_args.port == 12345
-    assert f.gateway_args.port == 12345
+    assert int(f.client_args.port) == 12345
+    assert int(f.gateway_args.port) == 12345
     f._update_network_interface(port=54321)
-    assert f.client_args.port == 54321
-    assert f.gateway_args.port == 54321
+    assert int(f.client_args.port) == 54321
+    assert int(f.gateway_args.port) == 54321
diff --git a/tests/unit/test_helper.py b/tests/unit/test_helper.py
index 942f81ca6e7c7..210c52d9b8318 100644
--- a/tests/unit/test_helper.py
+++ b/tests/unit/test_helper.py
@@ -9,6 +9,7 @@
 from jina.clients.helper import _safe_callback, pprint_routes
 from jina.excepts import BadClientCallback, NotSupportedError
 from jina.helper import (
+    _parse_hosts,
     _parse_ports,
     cached_property,
     convert_tuple_to_list,
@@ -18,6 +19,7 @@
     get_ci_vendor,
     is_port_free,
     is_yaml_filepath,
+    make_iterable,
     parse_arg,
     random_port,
     reset_ports,
@@ -370,6 +372,27 @@ def test_parse_port(port, output):
     assert _parse_ports(port) == output
 
 
+@pytest.mark.parametrize(
+    'host, output',
+    [('1234', '1234'), ('1,2,6', ['1', '2', '6'])],
+)
+def test_parse_host(host, output):
+    assert _parse_hosts(host) == output
+
+
+@pytest.mark.parametrize(
+    'o, expected',
+    [
+        (1, [1]),
+        ('a', ['a']),
+        (bytes('hello', 'utf-8'), [bytes('hello', 'utf-8')]),
+        ([1, 2], [1, 2]),
+    ],
+)
+def test_make_iterable(o, expected):
+    assert make_iterable(o) == expected
+
+
 @pytest.mark.parametrize(
     'input, expected',
     [


Code review comments:
diff_hunk: @@ -339,5 +381,46 @@ def foo(self, docs, **kwargs):
         with f:
             response = f.post(on='/', inputs=d, target_executor='external_executor')
 
-
     assert response[0].text == 'external'
+
+
+@pytest.mark.parametrize('hosts', ['localhost,localhost', 'localhost'])
path: tests/integration/external_deployment/test_external_deployment.py
text: @user1:
External deployment and external replicas should be considered 2 different features and in my opinion tested differently in different test folders

@author:
ok will refactor

diff_hunk: @@ -1181,6 +1195,20 @@ def add(
                 'External Executors with multiple needs have to do auto reduce.'
             )
 
+        # args.external_replica_ports = make_iterable(_parse_ports(str(args.port)))
+        # args.external_replica_hosts = make_iterable(_parse_hosts(str(args.host)))
path: jina/orchestrate/flow/base.py
text: @user1:
remember to remve if needed

diff_hunk: @@ -0,0 +1,7 @@
+FROM jinaai/jina:master
path: tests/integration/external_deployment/Dockerfile
text: @user1:
```suggestion
FROM jinaai/jina:test-pip
```