Repository: jina-ai/serve
Pull request: feat: allow to pass a list of port monitoring to replicas 
# Context:

Right now it is difficult to use the monitoring feature with replicas/shards because it is impossible to gave a list of port to the Deployment. The current behavior is to randomly assign the port when there is a replicas. It is actually the same behavior as for the normal port. But since the `port_monitoring` should be access by Prometheus that is external we can't only rely on this behavior.

It would be great if we could pass several port to a Deployment.

# What this PR do:

Allow to pass a list of port to the Deployment. The ports should be separated by a comma.

```python
from jina import Flow, Executor

with Flow(monitoring=True).add(uses=Executor, replicas=2, port_monitoring='9092,9091'):
   ...
```


- [x] Allow to pass a list
- [x] Add tests
- [x] Add documentation

Problem statement:
Monitoring: port issue when using shards and replicas locally
There is an port conflict when using shards and replicas


```python
from jina import Flow
from docarray import Document, DocumentArray

with Flow(monitoring=True).add(
    name='hello', port_monitoring=9090, replicas=2, shards=2
) as f:
    # f.index(inputs=DocumentArray.empty(size=5))
    f.block()
```


```bash
ERROR  hello/head@206589 OSError(98, 'Address already in     [06/24/22 13:04:49]
       use') during <class                                                      
       'jina.serve.runtimes.head.HeadRuntime'>                                  
       initialization                                                           
        add "--quiet-error" to suppress the exception                           
       details                                                                  
       ╭──────── Traceback (most recent call last) ────────╮                    
       │ /home/sami/Documents/workspace/Jina/jina/jina/or… │                    
       │ in run                                            │                    
       │                                                   │                    
       │    71 │                                           │                    
       │    72 │   try:                                    │                    
       │    73 │   │   _set_envs()                         │                    
       │ ❱  74 │   │   runtime = runtime_cls(              │                    
       │    75 │   │   │   args=args,                      │                    
       │    76 │   │   )                                   │                    
       │    77 │   except Exception as ex:                 │                    
       │                                                   │                    
       │ /home/sami/Documents/workspace/Jina/jina/jina/se… │                    
       │ in __init__                                       │                    
       │                                                   │                    
       │    40 │   │   """                                 │                    
       │    41 │   │   self._health_servicer = health.Heal │                    
       │    42 │   │                                       │                    
       │ ❱  43 │   │   super().__init__(args, **kwargs)    │                    
       │    44 │   │   if args.name is None:               │                    
       │    45 │   │   │   args.name = ''                  │                    
       │    46 │   │   self.name = args.name               │                    
       │                                                   │                    
       │ /home/sami/Documents/workspace/Jina/jina/jina/se… │                    
       │ in __init__                                       │                    
       │                                                   │                    
       │    62 │   │   │   │   lambda *args, **kwargs: sel │                    
       │    63 │   │   │   )                               │                    
       │    64 │   │                                       │                    
       │ ❱  65 │   │   self._setup_monitoring()            │                    
       │    66 │   │   self._loop.run_until_complete(self. │                    
       │    67 │                                           │                    
       │    68 │   def run_forever(self):                  │                    
       │                                                   │                    
       │ /home/sami/Documents/workspace/Jina/jina/jina/se… │                    
       │ in _setup_monitoring                              │                    
       │                                                   │                    
       │   17 │   │   │                                    │                    
       │   18 │   │   │   from prometheus_client import st │                    
       │   19 │   │   │                                    │                    
       │ ❱ 20 │   │   │   start_http_server(self.args.port │                    
       │   21                                              │                    
       │                                                   │                    
       │ /home/sami/.pyenv/versions/jina3/lib/python3.9/s… │                    
       │ in start_wsgi_server                              │                    
       │                                                   │                    
       │   141 def start_wsgi_server(port, addr='', regist │                    
       │   142 │   """Starts a WSGI server for prometheus  │                    
       │   143 │   app = make_wsgi_app(registry)           │                    
       │ ❱ 144 │   httpd = make_server(addr, port, app, Th │                    
       │       handler_class=_SilentHandler)               │                    
       │   145 │   t = threading.Thread(target=httpd.serve │                    
       │   146 │   t.daemon = True                         │                    
       │   147 │   t.start()                               │                    
       │                                                   │                    
       │ /home/sami/.pyenv/versions/3.9.10/lib/python3.9/… │                    
       │ in make_server                                    │                    
       │                                                   │                    
       │   151 │   host, port, app, server_class=WSGIServe │                    
       │   152 ):                                          │                    
       │   153 │   """Create a new WSGI server listening o │                    
       │ ❱ 154 │   server = server_class((host, port), han │                    
       │   155 │   server.set_app(app)                     │                    
       │   156 │   return server                           │                    
       │   157                                             │                    
       │                                                   │                    
       │ /home/sami/.pyenv/versions/3.9.10/lib/python3.9/… │                    
       │ in __init__                                       │                    
       │                                                   │                    
       │   449 │   │   │   │   │   │   │   │   │   self.so │                    
       │   450 │   │   if bind_and_activate:               │                    
       │   451 │   │   │   try:                            │                    
       │ ❱ 452 │   │   │   │   self.server_bind()          │                    
       │   453 │   │   │   │   self.server_activate()      │                    
       │   454 │   │   │   except:                         │                    
       │   455 │   │   │   │   self.server_close()         │                    
       │                                                   │                    
       │ /home/sami/.pyenv/versions/3.9.10/lib/python3.9/… │                    
       │ in server_bind                                    │                    
       │                                                   │                    
       │    47 │                                           │                    
       │    48 │   def server_bind(self):                  │                    
       │    49 │   │   """Override server_bind to store th │                    
       │ ❱  50 │   │   HTTPServer.server_bind(self)        │                    
       │    51 │   │   self.setup_environ()                │                    
       │    52 │                                           │                    
       │    53 │   def setup_environ(self):                │                    
       │                                                   │                    
       │ /home/sami/.pyenv/versions/3.9.10/lib/python3.9/… │                    
       │ in server_bind                                    │                    
       │                                                   │                    
       │    135 │                                          │                    
       │    136 │   def server_bind(self):                 │                    
       │    137 │   │   """Override server_bind to store t │                    
       │ ❱  138 │   │   socketserver.TCPServer.server_bind │                    
       │    139 │   │   host, port = self.server_address[: │                    
       │    140 │   │   self.server_name = socket.getfqdn( │                    
       │    141 │   │   self.server_port = port            │                    
       │                                                   │                    
       │ /home/sami/.pyenv/versions/3.9.10/lib/python3.9/… │                    
       │ in server_bind                                    │                    
       │                                                   │                    
       │   463 │   │   """                                 │                    
       │   464 │   │   if self.allow_reuse_address:        │                    
       │   465 │   │   │   self.socket.setsockopt(socket.S │                    
       │ ❱ 466 │   │   self.socket.bind(self.server_addres │                    
       │   467 │   │   self.server_address = self.socket.g │                    
       │   468 │                                           │                    
       │   469 │   def server_activate(self):              │                    
       ╰───────────────────────────────────────────────────╯                    
       OSError: [Errno 98] Address already in use                               
ERROR  Flow@206580 Flow is aborted due to ['hello'] can not  [06/24/22 13:04:52]
       be started.                                                              

Traceback (most recent call last):
  File "/home/sami/.config/JetBrains/PyCharmCE2022.1/scratches/scratch_3.py", line 4, in <module>
    with Flow(monitoring=True).add(
  File "/home/sami/Documents/workspace/Jina/jina/jina/orchestrate/flow/base.py", line 1393, in __enter__
    return self.start()
  File "/home/sami/Documents/workspace/Jina/jina/jina/orchestrate/flow/builder.py", line 33, in arg_wrapper
    return func(self, *args, **kwargs)
  File "/home/sami/Documents/workspace/Jina/jina/jina/orchestrate/flow/base.py", line 1449, in start
    self._wait_until_all_ready()
  File "/home/sami/Documents/workspace/Jina/jina/jina/orchestrate/flow/base.py", line 1547, in _wait_until_all_ready
    raise RuntimeFailToStart
jina.excepts.RuntimeFailToStart
```

Hints from issue discussion:

Commit message: refactor: raname port to port monitoring and parser args

Patch to review:
diff --git a/docs/fundamentals/flow/monitoring-flow.md b/docs/fundamentals/flow/monitoring-flow.md
index ee72a229bb90f..00a544dde4df6 100644
--- a/docs/fundamentals/flow/monitoring-flow.md
+++ b/docs/fundamentals/flow/monitoring-flow.md
@@ -91,6 +91,51 @@ On the other hand, If you want to only enable the monitoring on a given Executor
 Flow().add(...).add(uses=MyExecutor, monitoring=True)
 ```
 
+### Enable monitoring with replicas and shards
+
+When using replicas or shards (see this how-to scale-out) all of the duplicate Executor will expose their own metrics.
+
+When deploying a Flow locally (i.e without Kubernetes or Docker Compose) all the duplicate will use a random `port_monitoring`
+expect one that will use the `port_monitoring` that you precise. You can as well pass a list of ports to be used by the different replicas 
+for the monitoring by passing a string of several ports separated by a comma.
+
+For example :
+
+````{tab} via Python API
+
+```python
+from jina import Flow
+
+with Flow(monitoring=True).add(
+    uses='jinahub://SimpleIndexer', replicas=2, port_monitoring='9091,9092'
+) as f:
+    f.block()
+```
+````
+
+````{tab} via YAML
+This example shows how to start a Flow with monitoring enabled via yaml:
+
+In a `flow.yaml` file
+```yaml
+jtype: Flow
+with:
+  monitoring: true
+executors:
+- uses: jinahub://SimpleIndexer
+  replicas=2
+  port_monitoring: '9091,9092'
+```
+
+```bash
+jina flow --uses flow.yaml
+```
+````
+
+```{tip} Monitoring with shards
+When using shards, an extra head will be created and you will need to pass a list of N+1 ports to `port_monitoring`, N beeing the number of shards you desire
+```
+
 ## Available metrics
 
 A {class}`~jina.Flow` supports different metrics out of the box, in addition to allowing the user to define their own custom metrics.
diff --git a/jina/helper.py b/jina/helper.py
index 0ced2d1f874c7..beea3b576d944 100644
--- a/jina/helper.py
+++ b/jina/helper.py
@@ -1618,3 +1618,30 @@ def is_port_free(host, port):
             return False
         else:
             return True
+
+
+def _parse_port(port: str) -> Union[int, List]:
+    """Parse port
+
+    EXAMPLE USAGE
+
+    .. code-block:: python
+
+
+        _parse_port('8000')
+        8000
+
+        _parse_port('8001:8002:8005')
+        [80001, 8002, 8005]
+
+    :param port: the string to parse
+    :return: the port or the iterable ports
+    """
+    try:
+        port = int(port)
+    except ValueError as e:
+        if ',' in port:
+            port = [int(port_) for port_ in port.split(',')]
+        else:
+            raise e
+    return port
diff --git a/jina/orchestrate/deployments/__init__.py b/jina/orchestrate/deployments/__init__.py
index 2a017a0e91436..d2fdfaf0b834c 100644
--- a/jina/orchestrate/deployments/__init__.py
+++ b/jina/orchestrate/deployments/__init__.py
@@ -12,7 +12,7 @@
 
 from jina import __default_executor__, __default_host__, __docker_host__, helper
 from jina.enums import DeploymentRoleType, PodRoleType, PollingType
-from jina.helper import CatchAllCleanupContextManager
+from jina.helper import CatchAllCleanupContextManager, _parse_port
 from jina.hubble.helper import replace_secret_of_hub_uri
 from jina.hubble.hubio import HubIO
 from jina.jaml.helper import complete_path
@@ -101,6 +101,13 @@ def head_port(self):
         """
         return self.head_args.port if self.head_args else None
 
+    @property
+    def head_port_monitoring(self):
+        """Get the port_monitoring of the HeadPod of this deployment
+        .. # noqa: DAR201
+        """
+        return self.head_args.port_monitoring if self.head_args else None
+
     def __enter__(self) -> 'BaseDeployment':
         with CatchAllCleanupContextManager(self):
             return self.start()
@@ -248,6 +255,7 @@ def __init__(
         self.uses_after_pod = None
         self.head_pod = None
         self.shards = {}
+        self._update_port_args()
         self.update_pod_args()
         self._sandbox_deployed = False
 
@@ -255,6 +263,17 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         super().__exit__(exc_type, exc_val, exc_tb)
         self.join()
 
+    def _update_port_args(self):
+        _all_port_monitoring = _parse_port(self.args.port_monitoring)
+        self.args.all_port_monitoring = (
+            [_all_port_monitoring]
+            if not type(_all_port_monitoring) == list
+            else _all_port_monitoring
+        )
+        self.args.port_monitoring = int(
+            self.args.all_port_monitoring[0]
+        )  # this is for the head
+
     def update_pod_args(self):
         """Update args of all its pods based on Deployment args. Including head/tail"""
         if isinstance(self.args, Dict):
@@ -688,9 +707,30 @@ def _set_pod_args(args: Namespace) -> Dict[int, List[Namespace]]:
                 if args.deployment_role == DeploymentRoleType.GATEWAY or args.external:
                     _args.port = args.port
                 elif args.shards == 1 and args.replicas == 1:
+                    _args.port = args.port
+                    _args.port_monitoring = args.port_monitoring
+
+                elif args.shards == 1:
+                    _args.port_monitoring = (
+                        helper.random_port()
+                        if replica_id >= len(args.all_port_monitoring)
+                        else args.all_port_monitoring[replica_id]
+                    )
                     # if there are no shards/replicas, we dont need to distribute ports randomly
                     # we should rather use the pre assigned one
-                    args.port = args.port
+                    _args.port = helper.random_port()
+                elif args.shards > 1:
+                    port_monitoring_index = (
+                        replica_id + args.replicas * shard_id + 1
+                    )  # the first index is for the head
+                    _args.port_monitoring = (
+                        helper.random_port()
+                        if port_monitoring_index >= len(args.all_port_monitoring)
+                        else args.all_port_monitoring[
+                            port_monitoring_index
+                        ]  # we skip the head port here
+                    )
+                    _args.port = helper.random_port()
                 else:
                     _args.port = helper.random_port()
                     _args.port_monitoring = helper.random_port()
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index eab83dca52e22..e10d3a4143576 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -798,7 +798,7 @@ def add(
         output_array_type: Optional[str] = None,
         polling: Optional[str] = 'ANY',
         port: Optional[int] = None,
-        port_monitoring: Optional[int] = None,
+        port_monitoring: Optional[str] = None,
         py_modules: Optional[List[str]] = None,
         quiet: Optional[bool] = False,
         quiet_error: Optional[bool] = False,
diff --git a/jina/parsers/__init__.py b/jina/parsers/__init__.py
index e02b58cf59489..8f1ad1397f968 100644
--- a/jina/parsers/__init__.py
+++ b/jina/parsers/__init__.py
@@ -3,10 +3,11 @@
 from jina.parsers.orchestrate.runtimes.head import mixin_head_parser
 
 
-def set_pod_parser(parser=None):
+def set_pod_parser(parser=None, port_monitoring=True):
     """Set the parser for the Pod
 
     :param parser: an optional existing parser to build upon
+    :param port_monitoring: if to include the port parsing
     :return: the parser
     """
     if not parser:
@@ -31,7 +32,7 @@ def set_pod_parser(parser=None):
     mixin_container_runtime_parser(parser)
     mixin_remote_runtime_parser(parser)
     mixin_distributed_feature_parser(parser)
-    mixin_pod_parser(parser)
+    mixin_pod_parser(parser, port_monitoring=port_monitoring)
     mixin_hub_pull_options_parser(parser)
     mixin_head_parser(parser)
 
@@ -49,7 +50,7 @@ def set_deployment_parser(parser=None):
 
         parser = set_base_parser()
 
-    set_pod_parser(parser)
+    set_pod_parser(parser, port_monitoring=False)
 
     from jina.parsers.orchestrate.deployment import mixin_base_deployment_parser
 
diff --git a/jina/parsers/orchestrate/deployment.py b/jina/parsers/orchestrate/deployment.py
index 45f6fed5d24af..3ef6feda57772 100644
--- a/jina/parsers/orchestrate/deployment.py
+++ b/jina/parsers/orchestrate/deployment.py
@@ -1,6 +1,7 @@
 """Argparser module for Deployment runtimes"""
 import argparse
 
+from jina import helper
 from jina.enums import DeploymentRoleType
 from jina.parsers.helper import _SHOW_ALL_ARGS, KVAppendAction, add_arg_group
 
@@ -59,3 +60,11 @@ def mixin_base_deployment_parser(parser):
         default=False,
         help='If set, connect to deployment using tls encryption',
     )
+
+    gp.add_argument(
+        '--port-monitoring',
+        type=str,
+        default=str(helper.random_port()),
+        dest='port_monitoring',
+        help=f'The port on which the prometheus server is exposed, default is a random port between [49152, 65535]',
+    )
diff --git a/jina/parsers/orchestrate/pod.py b/jina/parsers/orchestrate/pod.py
index 4b0bd71a80c08..2ebf7c348224b 100644
--- a/jina/parsers/orchestrate/pod.py
+++ b/jina/parsers/orchestrate/pod.py
@@ -6,9 +6,10 @@
 from jina.parsers.helper import _SHOW_ALL_ARGS, KVAppendAction, add_arg_group
 
 
-def mixin_pod_parser(parser):
+def mixin_pod_parser(parser, port_monitoring=True):
     """Mixing in arguments required by :class:`Pod` into the given parser.
     :param parser: the parser instance to which we add arguments
+    :param port_monitoring: if to include the port parsing
     """
 
     gp = add_arg_group(parser, title='Pod')
@@ -96,13 +97,14 @@ def mixin_pod_parser(parser):
         help='If set, spawn an http server with a prometheus endpoint to expose metrics',
     )
 
-    gp.add_argument(
-        '--port-monitoring',
-        type=int,
-        default=helper.random_port(),  # default prometheus server port
-        dest='port_monitoring',
-        help=f'The port on which the prometheus server is exposed, default is a random port between [49152, 65535]',
-    )
+    if port_monitoring:
+        gp.add_argument(
+            '--port-monitoring',
+            type=int,
+            default=helper.random_port(),
+            dest='port_monitoring',
+            help=f'The port on which the prometheus server is exposed, default is a random port between [49152, 65535]',
+        )
 
     gp.add_argument(
         '--retries',
diff --git a/jina_cli/autocomplete.py b/jina_cli/autocomplete.py
index eb2b9f04740d8..d35c47c68484f 100644
--- a/jina_cli/autocomplete.py
+++ b/jina_cli/autocomplete.py
@@ -268,7 +268,6 @@
             '--replicas',
             '--port',
             '--monitoring',
-            '--port-monitoring',
             '--retries',
             '--floating',
             '--install-requirements',
@@ -286,6 +285,7 @@
             '--external',
             '--deployment-role',
             '--tls',
+            '--port-monitoring',
         ],
         'client': [
             '--help',
diff --git a/tests/integration/monitoring/test_monitoring.py b/tests/integration/monitoring/test_monitoring.py
index 174206dd86acd..7a9c9dc91f3c7 100644
--- a/tests/integration/monitoring/test_monitoring.py
+++ b/tests/integration/monitoring/test_monitoring.py
@@ -62,22 +62,94 @@ def test_enable_monitoring_gateway(protocol, port_generator, executor):
 
 
 def test_monitoring_head(port_generator, executor):
+    n_shards = 2
+    port_shards_list = [port_generator() for _ in range(n_shards)]
+    port_head = port_generator()
+    port_monitoring = ','.join([str(port) for port in [port_head] + port_shards_list])
+    port1 = port_generator()
+
+    f = Flow(monitoring=True, port_monitoring=port1).add(
+        uses=executor, port_monitoring=port_monitoring, shards=n_shards
+    )
+
+    assert f._deployment_nodes['executor0'].head_port_monitoring == port_head
+
+    unique_port_exposed = set(
+        [
+            pod[0].port_monitoring
+            for key, pod in f._deployment_nodes['executor0'].pod_args['pods'].items()
+        ]
+    )
+
+    assert unique_port_exposed == set(port_shards_list)
+    with f:
+        for port in [port_head, port1] + port_shards_list:
+            resp = req.get(f'http://localhost:{port}/')
+            assert resp.status_code == 200
+
+        f.search(inputs=DocumentArray())
+        resp = req.get(f'http://localhost:{port_head}/')
+        assert f'jina_receiving_request_seconds' in str(resp.content)
+        assert f'jina_sending_request_seconds' in str(resp.content)
+
+
+def test_monitoring_head_few_port(port_generator, executor):
+    n_shards = 2
     port1 = port_generator()
     port2 = port_generator()
 
-    with Flow(monitoring=True, port_monitoring=port_generator()).add(
-        uses=executor, port_monitoring=port1
-    ).add(uses=executor, port_monitoring=port2, shards=2) as f:
+    f = Flow(monitoring=True, port_monitoring=port1).add(
+        uses=executor, port_monitoring=port2, shards=n_shards
+    )
 
-        port3 = f._deployment_nodes['executor0'].pod_args['pods'][0][0].port_monitoring
-        port4 = f._deployment_nodes['executor1'].pod_args['pods'][0][0].port_monitoring
+    assert f._deployment_nodes['executor0'].head_port_monitoring == port2
 
-        for port in [port1, port2, port3, port4]:
+    unique_port_exposed = set(
+        [
+            pod[0].port_monitoring
+            for key, pod in f._deployment_nodes['executor0'].pod_args['pods'].items()
+        ]
+    )
+    with f:
+        for port in [port2, port1]:
+            resp = req.get(f'http://localhost:{port}/')
+            assert resp.status_code == 200
+
+
+def test_monitoring_replicas_and_shards(port_generator, executor):
+    n_shards = 2
+    n_replicas = 3
+    port_shards_list = [port_generator() for _ in range(n_shards * n_replicas)]
+    port_head = port_generator()
+    port_monitoring = ','.join([str(port) for port in [port_head] + port_shards_list])
+    port1 = port_generator()
+
+    f = Flow(monitoring=True, port_monitoring=port1).add(
+        uses=executor,
+        port_monitoring=port_monitoring,
+        shards=n_shards,
+        replicas=n_replicas,
+    )
+
+    assert f._deployment_nodes['executor0'].head_port_monitoring == port_head
+
+    unique_port_exposed = set(
+        [
+            pod.port_monitoring
+            for _, deployment in f._deployment_nodes.items()
+            for _, list_pod in deployment.pod_args['pods'].items()
+            for pod in list_pod
+        ]
+    )
+
+    assert unique_port_exposed == set(port_shards_list)
+    with f:
+        for port in [port_head, port1] + port_shards_list:
             resp = req.get(f'http://localhost:{port}/')
             assert resp.status_code == 200
 
         f.search(inputs=DocumentArray())
-        resp = req.get(f'http://localhost:{port2}/')
+        resp = req.get(f'http://localhost:{port_head}/')
         assert f'jina_receiving_request_seconds' in str(resp.content)
         assert f'jina_sending_request_seconds' in str(resp.content)
 
@@ -89,7 +161,6 @@ def test_document_processed_total(port_generator, executor):
     with Flow(monitoring=True, port_monitoring=port0).add(
         uses=executor, port_monitoring=port1
     ) as f:
-
         resp = req.get(f'http://localhost:{port1}/')
         assert resp.status_code == 200
 
@@ -162,7 +233,6 @@ def test_requests_size(port_generator, executor):
     with Flow(monitoring=True, port_monitoring=port0).add(
         uses=executor, port_monitoring=port1
     ) as f:
-
         f.post('/foo', inputs=DocumentArray.empty(size=1))
 
         resp = req.get(f'http://localhost:{port1}/')  # enable on port0
@@ -250,3 +320,28 @@ def _assert_pending_value(val: str):
             assert not code
 
         _assert_after()
+
+
+def test_monitoring_replicas_parsing(port_generator, executor):
+    n_replicas = 3
+    port_monitoring_list = [port_generator() for _ in range(n_replicas)]
+    port_monitoring = ','.join([str(port) for port in port_monitoring_list])
+
+    with Flow().add(
+        uses=executor,
+        port_monitoring=port_monitoring,
+        replicas=n_replicas,
+        monitoring=True,
+    ) as f:
+        unique_port_exposed = set(
+            [
+                pod.port_monitoring
+                for pod in f._deployment_nodes['executor0'].pod_args['pods'][0]
+            ]
+        )
+
+        assert unique_port_exposed == set(port_monitoring_list)
+
+        for port in port_monitoring_list:
+            resp = req.get(f'http://localhost:{port}/')
+            assert resp.status_code == 200
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow.py b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
index 0c71749b0553b..f546bcc21e50a 100644
--- a/tests/unit/orchestrate/flow/flow-construct/test_flow.py
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
@@ -608,3 +608,10 @@ def _validate_flow(f):
         else:
             for n in pod.needs:
                 assert 'end-gateway' in graph_dict[n]
+
+
+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
+        f.index(inputs=[])
diff --git a/tests/unit/test_helper.py b/tests/unit/test_helper.py
index 41e749c0d2d19..6a8804b79b378 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_port,
     cached_property,
     convert_tuple_to_list,
     deprecated_alias,
@@ -364,3 +365,11 @@ async def dummy():
 
     end_fd_count = p.num_fds()
     assert first_fd_count == end_fd_count
+
+
+@pytest.mark.parametrize(
+    'port, output',
+    [('8080', 8080), ('1,2,6', [1, 2, 6])],
+)
+def test_parse_port(port, output):
+    assert _parse_port(port) == output


Code review comments:
diff_hunk: @@ -1618,3 +1618,30 @@ def is_port_free(host, port):
             return False
         else:
             return True
+
+
+def _parse_port(port: str) -> Union[int, List]:
path: jina/helper.py
text: @user1:
```suggestion
def _parse_ports(port: str) -> Union[int, List]:
```
i think this should be plural

@author:
will do

diff_hunk: @@ -91,6 +91,51 @@ On the other hand, If you want to only enable the monitoring on a given Executor
 Flow().add(...).add(uses=MyExecutor, monitoring=True)
 ```
 
+### Enable monitoring with replicas and shards
+
+When using replicas or shards (see this how-to scale-out) all of the duplicate Executor will expose their own metrics.
path: docs/fundamentals/flow/monitoring-flow.md
text: @user1:
This doesn't look like a proper internal link

@author:
Oh that's because I forgot to add the link