Repository: jina-ai/serve
Pull request: fix: check if port is occupied
fix https://github.com/jina-ai/jina/issues/4555

Problem statement:
Two jina flow can run on the same port
**Describe the bug**

You can launch two flow on the same port, meaning two gateway serving on the same port without any of the two complaining.


**Step to reproduce**

In `toy.py`
```python
from jina import Flow, requests, Executor

class Dummy(Executor):

    @requests
    def foo(self,docs,**kwargs):
        ...

if __name__ == '__main__':

    with Flow(port=8888).add(uses=Dummy) as f:
        f.block()
```

Then do two times in two terminal

```bash
python toy.py
```
None of them complain, the second call should say that the port is already used

**Environment**
jina version 3.2.9

Hints from issue discussion:
let's give it a higher priority, i think it can be really confusing for users, especially multiple-same-port-gateway gives weird empty results, as we shown in CLIP-as-service

Commit message: refactor(tests): move test to issues

Patch to review:
diff --git a/jina/excepts.py b/jina/excepts.py
index 2f2e2e8a6dde7..4568b81f3ddf5 100644
--- a/jina/excepts.py
+++ b/jina/excepts.py
@@ -81,3 +81,7 @@ class DockerVersionError(SystemError, BaseJinaException):
 
 class NoContainerizedError(Exception, BaseJinaException):
     """Raised when trying to use non-containerized Executor in K8s or Docker Compose"""
+
+
+class PortAlreadyUsed(RuntimeError, BaseJinaException):
+    """Raised when to use a port which is already used"""
diff --git a/jina/helper.py b/jina/helper.py
index 45098d9587e13..af0e8d7de0036 100644
--- a/jina/helper.py
+++ b/jina/helper.py
@@ -15,6 +15,7 @@
 from collections.abc import MutableMapping
 from datetime import datetime
 from itertools import islice
+from socket import AF_INET, SOCK_STREAM, socket
 from types import SimpleNamespace
 from typing import (
     TYPE_CHECKING,
@@ -1626,3 +1627,11 @@ def _parse_url(host):
         port = None
 
     return scheme, host, port
+
+
+def is_port_free(host, port):
+    with socket(AF_INET, SOCK_STREAM) as session:
+        if session.connect_ex((host, port)) == 0:
+            return False
+        else:
+            return True
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 6d81e60b428a5..219549ec58d81 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -39,6 +39,7 @@
 from jina.excepts import (
     FlowMissingDeploymentError,
     FlowTopologyError,
+    PortAlreadyUsed,
     RuntimeFailToStart,
 )
 from jina.helper import (
@@ -51,6 +52,7 @@
     get_internal_ip,
     get_public_ip,
     get_rich_console,
+    is_port_free,
     typename,
 )
 from jina.jaml import JAMLCompatible
@@ -471,6 +473,12 @@ def _add_gateway(
 
         kwargs.update(self._common_kwargs)
         args = ArgNamespace.kwargs2namespace(kwargs, set_gateway_parser())
+
+        if not (
+            is_port_free(__default_host__, args.port)
+        ):  # we check if the port is not used at parsing time as well for robustness
+            raise PortAlreadyUsed(f'port:{args.port}')
+
         args.noblock_on_start = True
         args.expose_graphql_endpoint = (
             self.args.expose_graphql_endpoint
diff --git a/jina/serve/runtimes/gateway/grpc/__init__.py b/jina/serve/runtimes/gateway/grpc/__init__.py
index 8dde74f0d7feb..2b62ce4ddcf00 100644
--- a/jina/serve/runtimes/gateway/grpc/__init__.py
+++ b/jina/serve/runtimes/gateway/grpc/__init__.py
@@ -3,6 +3,8 @@
 import grpc
 
 from jina import __default_host__
+from jina.excepts import PortAlreadyUsed
+from jina.helper import is_port_free
 from jina.proto import jina_pb2_grpc
 from jina.serve.runtimes.gateway import GatewayRuntime
 from jina.serve.runtimes.gateway.request_handling import handle_request, handle_result
@@ -26,6 +28,9 @@ async def async_setup(self):
             os.unsetenv('http_proxy')
             os.unsetenv('https_proxy')
 
+        if not (is_port_free(__default_host__, self.args.port)):
+            raise PortAlreadyUsed(f'port:{self.args.port}')
+
         self.server = grpc.aio.server(
             options=[
                 ('grpc.max_send_message_length', -1),
diff --git a/tests/integration/gateway_clients/__init__.py b/tests/integration/gateway/gateway_clients/__init__.py
similarity index 100%
rename from tests/integration/gateway_clients/__init__.py
rename to tests/integration/gateway/gateway_clients/__init__.py
diff --git a/tests/integration/gateway_clients/test_clients_gateways.py b/tests/integration/gateway/gateway_clients/test_clients_gateways.py
similarity index 100%
rename from tests/integration/gateway_clients/test_clients_gateways.py
rename to tests/integration/gateway/gateway_clients/test_clients_gateways.py
diff --git a/tests/integration/gateway_clients/test_host_scheme.py b/tests/integration/gateway/gateway_clients/test_host_scheme.py
similarity index 100%
rename from tests/integration/gateway_clients/test_host_scheme.py
rename to tests/integration/gateway/gateway_clients/test_host_scheme.py
diff --git a/tests/integration/gateway_clients/test_streaming.py b/tests/integration/gateway/gateway_clients/test_streaming.py
similarity index 100%
rename from tests/integration/gateway_clients/test_streaming.py
rename to tests/integration/gateway/gateway_clients/test_streaming.py
index 4ace1fc46ad37..b537d9fbb3af9 100644
--- a/tests/integration/gateway_clients/test_streaming.py
+++ b/tests/integration/gateway/gateway_clients/test_streaming.py
@@ -1,22 +1,22 @@
-import time
-import os
 import asyncio
 import copy
 import multiprocessing
-import pytest
-
-from typing import Dict
+import os
+import time
 from datetime import datetime
 from functools import partial
+from typing import Dict
+
+import pytest
 
-from jina.parsers import set_gateway_parser
 from jina import Document, DocumentArray
+from jina.clients import Client
+from jina.helper import random_port
+from jina.parsers import set_gateway_parser
+from jina.serve import networking
 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 import networking
-from jina.helper import random_port
-from jina.clients import Client
 
 INPUT_LEN = 4
 INPUT_GEN_SLEEP_TIME = 1
diff --git a/tests/integration/gateway_non_blocking/__init__.py b/tests/integration/gateway/gateway_non_blocking/__init__.py
similarity index 100%
rename from tests/integration/gateway_non_blocking/__init__.py
rename to tests/integration/gateway/gateway_non_blocking/__init__.py
diff --git a/tests/integration/gateway_non_blocking/test_gateway_non_blocking.py b/tests/integration/gateway/gateway_non_blocking/test_gateway_non_blocking.py
similarity index 93%
rename from tests/integration/gateway_non_blocking/test_gateway_non_blocking.py
rename to tests/integration/gateway/gateway_non_blocking/test_gateway_non_blocking.py
index 55c33329cf1d1..f8a2edeb40087 100644
--- a/tests/integration/gateway_non_blocking/test_gateway_non_blocking.py
+++ b/tests/integration/gateway/gateway_non_blocking/test_gateway_non_blocking.py
@@ -1,8 +1,8 @@
-import pytest
+import time
 
-from jina import Flow, Executor, requests, DocumentArray, Document
+import pytest
 
-import time
+from jina import Document, DocumentArray, Executor, Flow, requests
 
 
 class FastSlowExecutor(Executor):
diff --git a/tests/integration/websocket_gateway/__init__.py b/tests/integration/gateway/websocket_gateway/__init__.py
similarity index 100%
rename from tests/integration/websocket_gateway/__init__.py
rename to tests/integration/gateway/websocket_gateway/__init__.py
diff --git a/tests/integration/websocket_gateway/test_subprotocols.py b/tests/integration/gateway/websocket_gateway/test_subprotocols.py
similarity index 100%
rename from tests/integration/websocket_gateway/test_subprotocols.py
rename to tests/integration/gateway/websocket_gateway/test_subprotocols.py
diff --git a/tests/integration/issues/github_4555/test_port.py b/tests/integration/issues/github_4555/test_port.py
new file mode 100644
index 0000000000000..11e17966c8bbc
--- /dev/null
+++ b/tests/integration/issues/github_4555/test_port.py
@@ -0,0 +1,21 @@
+import time
+from threading import Thread
+
+import pytest
+
+from jina import Flow
+from jina.excepts import PortAlreadyUsed
+
+
+def test_two_flow_using_on_same_port_failing(port_generator):
+    # hacky test but there is now good way to wait that a flow exit
+
+    port = port_generator()
+
+    f1 = Flow(port=port)
+    f2 = Flow(port=port)
+
+    with f1:
+        with pytest.raises(PortAlreadyUsed):
+            with f2:
+                pass
diff --git a/tests/unit/test_helper.py b/tests/unit/test_helper.py
index c1b2b0409b048..412a4b35c4687 100644
--- a/tests/unit/test_helper.py
+++ b/tests/unit/test_helper.py
@@ -4,7 +4,7 @@
 import numpy as np
 import pytest
 
-from jina import Executor, __default_endpoint__
+from jina import Executor, Flow, __default_endpoint__
 from jina.clients.helper import _safe_callback, pprint_routes
 from jina.excepts import BadClientCallback, NotSupportedError
 from jina.helper import (
@@ -14,6 +14,7 @@
     dunder_get,
     find_request_binding,
     get_ci_vendor,
+    is_port_free,
     is_yaml_filepath,
     random_port,
     reset_ports,
@@ -338,3 +339,16 @@ def try_it(self):
 
     assert result == 'it works'
     assert try_me.tried_count == 3
+
+
+def test_port_occupied(port_generator):
+    port = port_generator()
+    with Flow(port=port):
+        is_occupied = not (is_port_free('localhost', port))
+        assert is_occupied
+
+
+def test_port_free(port_generator):
+    port = port_generator()
+    is_free = is_port_free('localhost', port)
+    assert is_free


Code review comments:
diff_hunk: @@ -471,6 +473,12 @@ def _add_gateway(
 
         kwargs.update(self._common_kwargs)
         args = ArgNamespace.kwargs2namespace(kwargs, set_gateway_parser())
+
+        if not (
+            is_port_free(__default_host__, args.port)
path: jina/orchestrate/flow/base.py
text: @user1:
I don't think this is correct. You are checking if the port defined for the Flow on localhost is available. But the Flow will not necessarily run on localhost!
You may build a Flow on one machine, generate K8s configs and deploy them elsewhere. Your could would fail in this case when port is occupied on the building machine, which is wrong.

I think you rather need to check this when the Flow starts. I dont see how you can do this on build time as it is impossible to say where the Flow will be started.

@author:
Arf, you are completly right:

To add some context: I wanted initially to only check if the port is free before starting the grpc server. However because of some flakyness in the healt check it was not working proprely (there is still this check btw). Testing at parsing time is a hack see this pr for more info about the flakyness/bug : https://github.com/jina-ai/jina/pull/4565

I hack would be to check at parsing time only when run locally. 

The actual true solution is to fix the problem of the pr above, but we decide with @user2 that is was not worth it investigate as we move more and more to use k8s

@user1:
The issue with this is that it is basically impossible to know upfront if a Flow will run locally or not? You can't anticipate if the user will eventually start the Flow locally or generate some configs.
I think I dont really get the reasoning why you cant just move the check from build time to start time? If you just have this check on `Flow.start()` it should behave the same no? The only difference is that we can be sure then that the Flow is indeed running locally.

@user2:
yes, it should be moved to `start`

diff_hunk: @@ -1,22 +1,22 @@
-import time
-import os
 import asyncio
 import copy
 import multiprocessing
-import pytest
-
-from typing import Dict
+import os
path: tests/integration/gateway/gateway_clients/test_streaming.py
text: @user1:
these changes here are probably isort, but lets try to not change imports on files you are making no other changes?

@author:
That's strange that it has been caught in the pre-commit then :thinking: I will remove it