Repository: jina-ai/serve
Pull request: feat: Expose Executors as first class citizens
# Expose Executors as first class citizens

## Description

This adds  `to_k8s_yaml()` and `to_docker_compose_yaml()` for the Executors. It works similar to `serve()` in the sense that it creates a `Flow` behind the scenes and uses its existing methods to provide a similar user experience.
These features are also added to the docs and some basic tests have been added.

- [x] Implement `to_k8s_yaml()` and `to_docker_compose_yaml()` as a static convenience wrapper for Executors
- [x] Update the docs: Executors can now be deployed via the k8s/docker-compose mechanism. Explain the difference between external/shared Executors.

Closes [# (4502)](https://github.com/jina-ai/jina/issues/4502)

Problem statement:
Expose Executor as a Service as First class citizen in Documentation.
**Describe the feature**
Right now we have added `Executor.serve` method and we are seeing increasing interest in serving a single `Executor` as an isolated service.

After `Head` is removed as a result of the `service mesh` introduction, this will be even more performant, and `gateway` remains a useful entity in the picture. 

I think code-wise perspective, I belive the only needed thing is to add a method `Executor.to_k8s_yaml()` and `Executor.to_docker_compose_yaml` that under the hood create a simple `Flow` and call its `to_*` method.

The main work would be in documentation and storytelling about exposing this as a first-class option of `Jina`.

Hints from issue discussion:
Blocked by #4503 
Seems like a good thing to do.
I see some conceptual overlap with external Executors, and we probably need to be careful to make the distinction clear.
> Seems like a good thing to do. I see some conceptual overlap with external Executors, and we probably need to be careful to make the distinction clear.

Good point, did not think about it, but it would be the biggest blocker
Let's restrict this for now to gRPC? Otherwise it will be more complicatd to integrate as external Executor
> Let's restrict this for now to gRPC? Otherwise it will be more complicatd to integrate as external Executor

Makes sense. Remind me, did we land on `serve`d executors being the only recommended way of doing external Execs from now on?
I would say that `serve` is the recommended way. But this includes to_k8s_yaml/to_docker_compose methods @JoanFM was mentioning.
Also we may want to think about exposing `serve` as part of CLI? I think sth like `jina executor serve` or so makes sense? I dont to force people to run python code 
> Also we may want to think about exposing serve as part of CLI? 

Agree
We somehow should be able to deploy this with and without a gateway?

Commit message: feat: add k8s/docker export for Executors

Patch to review:
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 590150fb9a5db..62c2097caf316 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -1814,7 +1814,12 @@ def expose_endpoint(self, exec_endpoint: str, **kwargs):
     # for backward support
     join = needs
 
-    def to_k8s_yaml(self, output_base_path: str, k8s_namespace: Optional[str] = None):
+    def to_k8s_yaml(
+        self,
+        output_base_path: str,
+        k8s_namespace: Optional[str] = None,
+        include_gateway: Optional[bool] = True,
+    ):
         """
         Converts the Flow into a set of yaml deployments to deploy in Kubernetes.
 
@@ -1823,6 +1828,7 @@ def to_k8s_yaml(self, output_base_path: str, k8s_namespace: Optional[str] = None
 
         :param output_base_path: The base path where to dump all the yaml files
         :param k8s_namespace: The name of the k8s namespace to set for the configurations. If None, the name of the Flow will be used.
+        :param include_gateway: Defines if the gateway deployment should be included, defaults to True
         """
         import yaml
 
@@ -1834,7 +1840,7 @@ def to_k8s_yaml(self, output_base_path: str, k8s_namespace: Optional[str] = None
         k8s_namespace = k8s_namespace or self.args.name or 'default'
 
         for node, v in self._deployment_nodes.items():
-            if v.external:
+            if v.external or (node == 'gateway' and not include_gateway):
                 continue
             deployment_base = os.path.join(output_base_path, node)
             k8s_deployment = K8sDeploymentConfig(
@@ -1864,11 +1870,13 @@ def to_docker_compose_yaml(
         self,
         output_path: Optional[str] = None,
         network_name: Optional[str] = None,
+        include_gateway: Optional[bool] = True,
     ):
         """
         Converts the Flow into a yaml file to run with `docker-compose up`
         :param output_path: The output path for the yaml file
         :param network_name: The name of the network that will be used by the deployment name
+        :param include_gateway: Defines if the gateway deployment should be included, defaults to True
         """
         import yaml
 
@@ -1890,6 +1898,9 @@ def to_docker_compose_yaml(
         services = {}
 
         for node, v in self._deployment_nodes.items():
+            if v.external or (node == 'gateway' and not include_gateway):
+                continue
+
             docker_compose_deployment = DockerComposeConfig(
                 args=v.args,
                 deployments_addresses=self._get_docker_compose_deployments_addresses(),
diff --git a/jina/serve/executors/__init__.py b/jina/serve/executors/__init__.py
index 1f5e1256f5eaf..964676566ad37 100644
--- a/jina/serve/executors/__init__.py
+++ b/jina/serve/executors/__init__.py
@@ -9,6 +9,7 @@
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type, Union
 
 from jina import __args_executor_init__, __default_endpoint__
+from jina.enums import BetterEnum
 from jina.helper import (
     ArgNamespace,
     T,
@@ -359,6 +360,96 @@ def serve(
         with f:
             f.block(stop_event)
 
+    class StandaloneExecutorType(BetterEnum):
+        """
+        Type of standalone Executors
+        """
+
+        EXTERNAL = 0  # served by a gateway
+        SHARED = 1  # not served by a gateway, served by head/worker
+
+    @staticmethod
+    def to_k8s_yaml(
+        uses: str,
+        output_base_path: str,
+        k8s_namespace: Optional[str] = None,
+        executor_type: Optional[
+            StandaloneExecutorType
+        ] = StandaloneExecutorType.EXTERNAL,
+        uses_with: Optional[Dict] = None,
+        uses_metas: Optional[Dict] = None,
+        uses_requests: Optional[Dict] = None,
+        **kwargs,
+    ):
+        """
+        Converts the Executor into a set of yaml deployments to deploy in Kubernetes.
+
+        If you don't want to rebuild image on Jina Hub,
+        you can set `JINA_HUB_NO_IMAGE_REBUILD` environment variable.
+
+        :param uses: the Executor to use. Has to be containerized and accessible from K8s
+        :param output_base_path: The base path where to dump all the yaml files
+        :param k8s_namespace: The name of the k8s namespace to set for the configurations. If None, the name of the Flow will be used.
+        :param executor_type: The type of Executor. Can be external or shared. External Executors include the Gateway. Shared Executors don't. Defaults to External
+        :param uses_with: dictionary of parameters to overwrite from the default config's with field
+        :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
+        :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
+        :param kwargs: other kwargs accepted by the Flow, full list can be found `here <https://docs.jina.ai/api/jina.orchestrate.flow.base/>`
+        """
+        from jina import Flow
+
+        f = Flow(**kwargs).add(
+            uses=uses,
+            uses_with=uses_with,
+            uses_metas=uses_metas,
+            uses_requests=uses_requests,
+        )
+        f.to_k8s_yaml(
+            output_base_path=output_base_path,
+            k8s_namespace=k8s_namespace,
+            include_gateway=executor_type
+            == BaseExecutor.StandaloneExecutorType.EXTERNAL,
+        )
+
+    @staticmethod
+    def to_docker_compose_yaml(
+        uses: str,
+        output_path: Optional[str] = None,
+        network_name: Optional[str] = None,
+        executor_type: Optional[
+            StandaloneExecutorType
+        ] = StandaloneExecutorType.EXTERNAL,
+        uses_with: Optional[Dict] = None,
+        uses_metas: Optional[Dict] = None,
+        uses_requests: Optional[Dict] = None,
+        **kwargs,
+    ):
+        """
+        Converts the Executor into a yaml file to run with `docker-compose up`
+        :param uses: the Executor to use. Has to be containerized
+        :param output_path: The output path for the yaml file
+        :param network_name: The name of the network that will be used by the deployment name
+        :param executor_type: The type of Executor. Can be external or shared. External Executors include the Gateway. Shared Executors don't. Defaults to External
+        :param uses_with: dictionary of parameters to overwrite from the default config's with field
+        :param uses_metas: dictionary of parameters to overwrite from the default config's metas field
+        :param uses_requests: dictionary of parameters to overwrite from the default config's requests field
+        :param kwargs: other kwargs accepted by the Flow, full list can be found `here <https://docs.jina.ai/api/jina.orchestrate.flow.base/>`
+        """
+        from jina import Flow
+
+        f = Flow(**kwargs).add(
+            uses=uses,
+            uses_with=uses_with,
+            uses_metas=uses_metas,
+            uses_requests=uses_requests,
+        )
+        f.to_docker_compose_yaml(
+            output_path=output_path,
+            network_name=network_name,
+            include_gateway=executor_type
+            == BaseExecutor.StandaloneExecutorType.EXTERNAL,
+        )
+
 
 class ReducerExecutor(BaseExecutor):
     """
diff --git a/tests/unit/serve/executors/test_executor.py b/tests/unit/serve/executors/test_executor.py
index 26301be277df4..489b32a50a3db 100644
--- a/tests/unit/serve/executors/test_executor.py
+++ b/tests/unit/serve/executors/test_executor.py
@@ -4,6 +4,7 @@
 from unittest import mock
 
 import pytest
+import yaml
 from docarray import Document, DocumentArray
 
 from jina import Client, Executor, Flow, requests
@@ -366,3 +367,71 @@ def test_default_workspace(tmpdir):
         assert result_workspace == os.path.join(
             os.environ['JINA_DEFAULT_WORKSPACE_BASE'], 'WorkspaceExec', '0'
         )
+
+
+@pytest.mark.parametrize(
+    'exec_type',
+    [Executor.StandaloneExecutorType.EXTERNAL, Executor.StandaloneExecutorType.SHARED],
+)
+def test_to_k8s_yaml(tmpdir, exec_type):
+    Executor.to_k8s_yaml(
+        output_base_path=tmpdir,
+        port_expose=2020,
+        uses='jinahub+docker://DummyHubExecutor',
+        executor_type=exec_type,
+    )
+
+    with open(os.path.join(tmpdir, 'executor0', 'executor0.yml')) as f:
+        exec_yaml = list(yaml.safe_load_all(f))[-1]
+        assert exec_yaml['spec']['template']['spec']['containers'][0][
+            'image'
+        ].startswith('jinahub/')
+
+    if exec_type == Executor.StandaloneExecutorType.SHARED:
+        assert set(os.listdir(tmpdir)) == {
+            'executor0',
+        }
+    else:
+        assert set(os.listdir(tmpdir)) == {
+            'executor0',
+            'gateway',
+        }
+
+        with open(os.path.join(tmpdir, 'gateway', 'gateway.yml')) as f:
+            gatewayyaml = list(yaml.safe_load_all(f))[-1]
+            assert (
+                gatewayyaml['spec']['template']['spec']['containers'][0]['ports'][0][
+                    'containerPort'
+                ]
+                == 2020
+            )
+            gateway_args = gatewayyaml['spec']['template']['spec']['containers'][0][
+                'args'
+            ]
+            assert gateway_args[gateway_args.index('--port') + 1] == '2020'
+
+
+@pytest.mark.parametrize(
+    'exec_type',
+    [Executor.StandaloneExecutorType.EXTERNAL, Executor.StandaloneExecutorType.SHARED],
+)
+def test_to_docker_compose_yaml(tmpdir, exec_type):
+    compose_file = os.path.join(tmpdir, 'compose.yml')
+    Executor.to_docker_compose_yaml(
+        output_path=compose_file,
+        port_expose=2020,
+        uses='jinahub+docker://DummyHubExecutor',
+        executor_type=exec_type,
+    )
+
+    with open(compose_file) as f:
+        services = list(yaml.safe_load_all(f))[0]['services']
+        assert services['executor0']['image'].startswith('jinahub/')
+
+        if exec_type == Executor.StandaloneExecutorType.SHARED:
+            assert len(services) == 1
+        else:
+            assert len(services) == 2
+            assert services['gateway']['ports'][0] == '2020:2020'
+            gateway_args = services['gateway']['command']
+            assert gateway_args[gateway_args.index('--port') + 1] == '2020'


Code review comments:
diff_hunk: @@ -1814,7 +1814,12 @@ def expose_endpoint(self, exec_endpoint: str, **kwargs):
     # for backward support
     join = needs
 
-    def to_k8s_yaml(self, output_base_path: str, k8s_namespace: Optional[str] = None):
+    def to_k8s_yaml(
+        self,
+        output_base_path: str,
+        k8s_namespace: Optional[str] = None,
+        include_gateway: Optional[bool] = True,
path: jina/orchestrate/flow/base.py
text: @user1:
an optional boolean makes little sense, I would just put a proper default