Repository: jina-ai/serve
Pull request: feat: add GRPC metadata to Executors and Gateway deployments
<!---Decomposing the complex issue into subtasks can help you build it step-by-step. Thanks for your pull request! :rocket: --->
<!---We know that dev life is hectic, but **please provide a (brief) description** of what your PR does, and how it does it. **Otherwise, your PR cannot be reviewed!** --->
<!---This policy was agreed upon in a past company retro, and makes everyone's life a little easier. Thanks for your collaboration!--->

**Goals:**
<!---https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword--->

- resolves #5162 
- [x] add `deployments_metadata` arguments in gateway deployment
- [x] add `grpc_metadata` argument in the flow `.add()` function
- [x] supporting k8s and docker-compose exporting
- [x] unittest
- [x] integrate tests

Here, I use the clip-as-service as an real-world example to verify:

```python
from jina import Flow, DocumentArray

## replace your access token
TOKEN = '<TOKEN>'

da = DocumentArray.empty(10)
da.texts = ['hello world']* 10

with Flow().add(name='clip', host='api.clip.jina.ai', port=2096, tls=True, external=True, grpc_metadata={'authorization': TOKEN}) as f:
    r = f.post(on='/encode', inputs=da, request_size=8)
    print(r.embeddings.shape)

f.to_kubernetes_yaml('k8s_with_meta_deployment')
f.to_docker_compose_yaml('compose_with_meta_deployment.yml')
```

It should works and return the embeddings. And you can also check the resulted k8s and compose YAML files. 


- [x] check and update documentation. See [guide](https://github.com/jina-ai/jina/blob/master/CONTRIBUTING.md#-contributing-documentation) and ask the team.


Problem statement:
Define authentication credential for external executors in YAML
**Describe the feature**
<!-- A clear and concise description of what the feature is. -->

When the flow is using an external executor, and its access is protected with an authentication credential (i.g,. access token). For example, the hosted free clip-as-service `api.clip.jina.ai`. To access this service, the user must provide the access token, such as:

```python
from jina import Client

client = Client(host='api.clip.jina.ai', port=2096, tls=True, protocol='grpc')
client.post(on='/encode', metadata=(('authorization', '<your access token>'),))
```

Then, if we want to use this service as an external executor in a flow

```python
jtype: Flow
with:
  protocol: grpc
  port: 12345
executors:
  - name: encoder
    host: api.clip.jina.ai
    port: 2096
    external: true

  - name: indexer
    uses: jinahub://AnnLiteIndexer/v0.3.11
    uses_with:
      n_dim: 512
    workspace: workspace
    install_requirements: true
```
There is **no way to define the TOKEN in the flow**. 

**Your proposal**
<!-- copy past your code/pull request link -->

---
<!-- Optional, but really help us locate the problem faster -->

**Environment**
<!-- Run `jina --version-full` and copy paste the output here -->

**Screenshots**
<!-- If applicable, add screenshots to help explain your problem. -->

Hints from issue discussion:

Commit message: style: fix overload and cli autocomplete

Patch to review:
diff --git a/docs/fundamentals/flow/add-executors.md b/docs/fundamentals/flow/add-executors.md
index 4010a558dd37d..4e99deda6c258 100644
--- a/docs/fundamentals/flow/add-executors.md
+++ b/docs/fundamentals/flow/add-executors.md
@@ -128,6 +128,24 @@ Using `tls` to connect to the External Executor is especially needed to use an e
 for further details
 ```
 
+You can also pass the `grpc_metadata` parameter, which is a dictionary of key-value pairs to be passed along with the gRPC request to the external Executor. 
+An example is the `authorization` metadata, which is used to authenticate the request to the external Executor:
+
+```python
+from jina import Flow
+
+Flow().add(
+    host='123.45.67.89',
+    port=443,
+    external=True,
+    grpc_metadata={'authorization': '<TOKEN>'},
+)
+```
+
+```{hint}
+The `grpc_metadata` parameter here is the same as the `metadata` concept in gRPC. See the gRPC [documentation](https://grpc.io/docs/what-is-grpc/core-concepts/#metadata) for further details.
+```
+
 
 (floating-executors)=
 ## Floating Executors
diff --git a/docs/fundamentals/flow/gateway-args.md b/docs/fundamentals/flow/gateway-args.md
index 999129acfc4d7..8eefd85dad239 100644
--- a/docs/fundamentals/flow/gateway-args.md
+++ b/docs/fundamentals/flow/gateway-args.md
@@ -38,6 +38,7 @@
 | `graph_description` | Routing graph for the gateway | `string` | `{}` |
 | `graph_conditions` | Dictionary stating which filtering conditions each Executor in the graph requires to receive Documents. | `string` | `{}` |
 | `deployments_addresses` | dictionary JSON with the input addresses of each Deployment | `string` | `{}` |
+| `deployments_metadata` | dictionary JSON with the request metadata for each Deployment | `string` | `{}` |
 | `deployments_disable_reduce` | list JSON disabling the built-in merging mechanism for each Deployment listed | `string` | `[]` |
 | `compression` | The compression mechanism used when sending requests from the Head to the WorkerRuntimes. For more details, check https://grpc.github.io/grpc/python/grpc.html#compression. | `string` | `None` |
 | `timeout_send` | The timeout in milliseconds used when sending data requests to Executors, -1 means no timeout, disabled by default | `number` | `None` |
diff --git a/jina/orchestrate/deployments/__init__.py b/jina/orchestrate/deployments/__init__.py
index 6dfeddb648d50..641dec063841e 100644
--- a/jina/orchestrate/deployments/__init__.py
+++ b/jina/orchestrate/deployments/__init__.py
@@ -421,6 +421,14 @@ def external(self) -> bool:
         """
         return getattr(self.args, 'external', False) or self.is_sandbox
 
+    @property
+    def grpc_metadata(self):
+        """
+        Get the gRPC metadata for this deployment.
+        :return: The gRPC metadata for this deployment. If the deployment is a gateway, return None.
+        """
+        return getattr(self.args, 'grpc_metadata', None)
+
     @property
     def protocol(self):
         """
diff --git a/jina/orchestrate/deployments/config/k8s.py b/jina/orchestrate/deployments/config/k8s.py
index 27624a0a21775..f1e9f9ab03ed5 100644
--- a/jina/orchestrate/deployments/config/k8s.py
+++ b/jina/orchestrate/deployments/config/k8s.py
@@ -29,16 +29,17 @@ class K8sDeploymentConfig:
 
     class _K8sDeployment:
         def __init__(
-                self,
-                name: str,
-                version: str,
-                pod_type: PodRoleType,
-                jina_deployment_name: str,
-                shard_id: Optional[int],
-                common_args: Union['Namespace', Dict],
-                deployment_args: Union['Namespace', Dict],
-                k8s_namespace: str,
-                k8s_deployments_addresses: Optional[Dict[str, List[str]]] = None,
+            self,
+            name: str,
+            version: str,
+            pod_type: PodRoleType,
+            jina_deployment_name: str,
+            shard_id: Optional[int],
+            common_args: Union['Namespace', Dict],
+            deployment_args: Union['Namespace', Dict],
+            k8s_namespace: str,
+            k8s_deployments_addresses: Optional[Dict[str, List[str]]] = None,
+            k8s_deployments_metadata: Optional[Dict[str, Dict[str, str]]] = None,
         ):
             self.name = name
             self.dns_name = to_compatible_name(name)
@@ -51,6 +52,7 @@ def __init__(
             self.num_replicas = getattr(self.deployment_args, 'replicas', 1)
             self.k8s_namespace = k8s_namespace
             self.k8s_deployments_addresses = k8s_deployments_addresses
+            self.k8s_deployments_metadata = k8s_deployments_metadata
 
         def get_gateway_yamls(
                 self,
@@ -212,10 +214,11 @@ def get_runtime_yamls(
             )
 
     def __init__(
-            self,
-            args: Union['Namespace', Dict],
-            k8s_namespace: Optional[str] = None,
-            k8s_deployments_addresses: Optional[Dict[str, List[str]]] = None,
+        self,
+        args: Union['Namespace', Dict],
+        k8s_namespace: Optional[str] = None,
+        k8s_deployments_addresses: Optional[Dict[str, List[str]]] = None,
+        k8s_deployments_metadata: Optional[Dict[str, Dict[str, str]]] = None,
     ):
         # External Deployments should be ignored in a K8s based Flow
         assert not (hasattr(args, 'external') and args.external)
@@ -226,6 +229,7 @@ def __init__(
             )
         self.k8s_namespace = k8s_namespace
         self.k8s_deployments_addresses = k8s_deployments_addresses
+        self.k8s_deployments_metadata = k8s_deployments_metadata
         self.head_deployment = None
         self.args = copy.copy(args)
         if k8s_namespace is not None:
@@ -247,7 +251,6 @@ def __init__(
                 k8s_namespace=self.k8s_namespace,
                 k8s_deployments_addresses=self.k8s_deployments_addresses,
             )
-
         self.worker_deployments = []
         deployment_args = self.deployment_args['deployments']
         for i, args in enumerate(deployment_args):
@@ -267,6 +270,9 @@ def __init__(
                     k8s_deployments_addresses=self.k8s_deployments_addresses
                     if name == 'gateway'
                     else None,
+                    k8s_deployments_metadata=self.k8s_deployments_metadata
+                    if name == 'gateway'
+                    else None,
                 )
             )
 
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index 109a2722e3843..60a30f6340b99 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -161,6 +161,7 @@ def __init__(
         cors: Optional[bool] = False,
         deployments_addresses: Optional[str] = '{}',
         deployments_disable_reduce: Optional[str] = '[]',
+        deployments_metadata: Optional[str] = '{}',
         description: Optional[str] = None,
         disable_auto_volume: Optional[bool] = False,
         docker_kwargs: Optional[dict] = None,
@@ -221,6 +222,7 @@ def __init__(
         :param cors: If set, a CORS middleware is added to FastAPI frontend to allow cross-origin access.
         :param deployments_addresses: dictionary JSON with the input addresses of each Deployment
         :param deployments_disable_reduce: list JSON disabling the built-in merging mechanism for each Deployment listed
+        :param deployments_metadata: dictionary JSON with the request metadata for each Deployment
         :param description: The description of this HTTP server. It will be used in automatics docs such as Swagger UI.
         :param disable_auto_volume: Do not automatically mount a volume for dockerized Executors.
         :param docker_kwargs: Dictionary of kwargs arguments that will be passed to Docker SDK when starting the docker '
@@ -424,6 +426,7 @@ def __init__(
         :param cors: If set, a CORS middleware is added to FastAPI frontend to allow cross-origin access.
         :param deployments_addresses: dictionary JSON with the input addresses of each Deployment
         :param deployments_disable_reduce: list JSON disabling the built-in merging mechanism for each Deployment listed
+        :param deployments_metadata: dictionary JSON with the request metadata for each Deployment
         :param description: The description of this HTTP server. It will be used in automatics docs such as Swagger UI.
         :param disable_auto_volume: Do not automatically mount a volume for dockerized Executors.
         :param docker_kwargs: Dictionary of kwargs arguments that will be passed to Docker SDK when starting the docker '
@@ -662,6 +665,7 @@ def _add_gateway(
         needs: str,
         graph_description: Dict[str, List[str]],
         deployments_addresses: Dict[str, List[str]],
+        deployments_metadata: Dict[str, Dict[str, str]],
         graph_conditions: Dict[str, Dict],
         deployments_disabled_reduce: List[str],
         **kwargs,
@@ -690,9 +694,21 @@ def _add_gateway(
         args.graph_description = json.dumps(graph_description)
         args.graph_conditions = json.dumps(graph_conditions)
         args.deployments_addresses = json.dumps(deployments_addresses)
+        args.deployments_metadata = json.dumps(deployments_metadata)
         args.deployments_disable_reduce = json.dumps(deployments_disabled_reduce)
         self._deployment_nodes[GATEWAY_NAME] = Deployment(args, needs)
 
+    def _get_deployments_metadata(self) -> Dict[str, Dict[str, str]]:
+        """Get the metadata of all deployments in the Flow
+
+        :return: a dictionary of deployment name and its metadata
+        """
+        return {
+            name: deployment.grpc_metadata
+            for name, deployment in self._deployment_nodes.items()
+            if deployment.grpc_metadata
+        }
+
     def _get_deployments_addresses(self) -> Dict[str, List[str]]:
         graph_dict = {}
         for node, deployment in self._deployment_nodes.items():
@@ -750,6 +766,15 @@ def _get_k8s_deployments_addresses(
 
         return graph_dict if graph_dict else None
 
+    def _get_k8s_deployments_metadata(self) -> Dict[str, List[str]]:
+        graph_dict = {}
+
+        for node, v in self._deployment_nodes.items():
+            if v.grpc_metadata:
+                graph_dict[node] = v.grpc_metadata
+
+        return graph_dict or None
+
     def _get_docker_compose_deployments_addresses(self) -> Dict[str, List[str]]:
         graph_dict = {}
         from jina.orchestrate.deployments.config.docker_compose import port
@@ -884,6 +909,7 @@ def add(
         floating: Optional[bool] = False,
         force_update: Optional[bool] = False,
         gpus: Optional[str] = None,
+        grpc_metadata: Optional[dict] = None,
         grpc_server_options: Optional[dict] = None,
         host: Optional[str] = '0.0.0.0',
         host_in: Optional[str] = '0.0.0.0',
@@ -952,6 +978,7 @@ def add(
               - To access specified gpus based on device id, use `--gpus device=[YOUR-GPU-DEVICE-ID]`
               - To access specified gpus based on multiple device id, use `--gpus device=[YOUR-GPU-DEVICE-ID1],device=[YOUR-GPU-DEVICE-ID2]`
               - To specify more parameters, use `--gpus device=[YOUR-GPU-DEVICE-ID],runtime=nvidia,capabilities=display
+        :param grpc_metadata: The metadata to be passed to the gRPC request.
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
         :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
         :param host_in: The host address for binding to, by default it is 0.0.0.0
@@ -1107,6 +1134,7 @@ def add(
               - To access specified gpus based on device id, use `--gpus device=[YOUR-GPU-DEVICE-ID]`
               - To access specified gpus based on multiple device id, use `--gpus device=[YOUR-GPU-DEVICE-ID1],device=[YOUR-GPU-DEVICE-ID2]`
               - To specify more parameters, use `--gpus device=[YOUR-GPU-DEVICE-ID],runtime=nvidia,capabilities=display
+        :param grpc_metadata: The metadata to be passed to the gRPC request.
         :param grpc_server_options: Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1}
         :param host: The host address of the runtime, by default it is 0.0.0.0. In the case of an external Executor (`--external` or `external=True`) this can be a list of hosts, separated by commas. Then, every resulting address will be considered as one replica of the Executor.
         :param host_in: The host address for binding to, by default it is 0.0.0.0
@@ -1257,7 +1285,6 @@ def add(
                 deployment_role=deployment_role,
             )
         )
-
         parser = set_deployment_parser()
         if deployment_role == DeploymentRoleType.GATEWAY:
             parser = set_gateway_parser()
@@ -1452,6 +1479,7 @@ def build(
                 needs={op_flow._last_deployment},
                 graph_description=op_flow._get_graph_representation(),
                 deployments_addresses=op_flow._get_deployments_addresses(),
+                deployments_metadata=op_flow._get_deployments_metadata(),
                 graph_conditions=op_flow._get_graph_conditions(),
                 deployments_disabled_reduce=op_flow._get_disabled_reduce_deployments(),
                 uses=op_flow.args.uses,
@@ -2408,6 +2436,9 @@ def to_kubernetes_yaml(
                 )
                 if node == 'gateway'
                 else None,
+                k8s_deployments_metadata=self._get_k8s_deployments_metadata()
+                if node == 'gateway'
+                else None,
             )
             configs = k8s_deployment.to_kubernetes_yaml()
             for name, k8s_objects in configs:
diff --git a/jina/parsers/orchestrate/deployment.py b/jina/parsers/orchestrate/deployment.py
index 4aa0b6e1bbcd3..358c6cce963d1 100644
--- a/jina/parsers/orchestrate/deployment.py
+++ b/jina/parsers/orchestrate/deployment.py
@@ -43,6 +43,14 @@ def mixin_base_deployment_parser(parser):
         'This Deployment will not be context managed by the Flow.',
     )
 
+    gp.add_argument(
+        '--grpc-metadata',
+        action=KVAppendAction,
+        metavar='KEY: VALUE',
+        nargs='*',
+        help='The metadata to be passed to the gRPC request.',
+    )
+
     # hidden CLI used for internal only
 
     gp.add_argument(
diff --git a/jina/parsers/orchestrate/runtimes/remote.py b/jina/parsers/orchestrate/runtimes/remote.py
index 563511d3fe38f..d58ae5c19adca 100644
--- a/jina/parsers/orchestrate/runtimes/remote.py
+++ b/jina/parsers/orchestrate/runtimes/remote.py
@@ -118,6 +118,13 @@ def mixin_gateway_parser(parser):
         default='{}',
     )
 
+    parser.add_argument(
+        '--deployments-metadata',
+        type=str,
+        help='dictionary JSON with the request metadata for each Deployment',
+        default='{}',
+    )
+
     parser.add_argument(
         '--deployments-disable-reduce',
         type=str,
diff --git a/jina/serve/gateway.py b/jina/serve/gateway.py
index a4483a194ad8f..649c7c4afd960 100644
--- a/jina/serve/gateway.py
+++ b/jina/serve/gateway.py
@@ -113,12 +113,14 @@ def inject_dependencies(
         graph_description = json.loads(args.graph_description)
         graph_conditions = json.loads(args.graph_conditions)
         deployments_addresses = json.loads(args.deployments_addresses)
+        deployments_metadata = json.loads(args.deployments_metadata)
         deployments_disable_reduce = json.loads(args.deployments_disable_reduce)
 
         self.streamer = GatewayStreamer(
             graph_representation=graph_description,
             executor_addresses=deployments_addresses,
             graph_conditions=graph_conditions,
+            deployments_metadata=deployments_metadata,
             deployments_disable_reduce=deployments_disable_reduce,
             timeout_send=timeout_send,
             retries=args.retries,
diff --git a/jina/serve/networking.py b/jina/serve/networking.py
index a40c2a388e314..a4382a2b1b3b0 100644
--- a/jina/serve/networking.py
+++ b/jina/serve/networking.py
@@ -384,6 +384,7 @@ async def send_requests(
             if not self._initialized:
                 await self._init_stubs()
             request_type = type(requests[0])
+
             if request_type == DataRequest and len(requests) == 1:
                 if self.single_data_stub:
                     call_result = self.single_data_stub.process_single_data(
@@ -432,7 +433,10 @@ async def send_requests(
                         self._histograms.histogram_metric_labels,
                     ):
                         async for response in self.stream_stub.Call(
-                            iter(requests), compression=compression, timeout=timeout
+                            iter(requests),
+                            compression=compression,
+                            timeout=timeout,
+                            metadata=metadata,
                         ):
                             if self._metrics.received_response_bytes:
                                 self._metrics.received_response_bytes.observe(
@@ -743,6 +747,7 @@ def send_request(
         shard_id: Optional[int] = None,
         polling_type: PollingType = PollingType.ANY,
         endpoint: Optional[str] = None,
+        metadata: Optional[Dict[str, str]] = None,
         timeout: Optional[float] = None,
         retries: Optional[int] = -1,
     ) -> List[asyncio.Task]:
@@ -753,6 +758,7 @@ def send_request(
         :param shard_id: Send to a specific shard of the deployment, ignored for polling ALL
         :param polling_type: defines if the message should be send to any or all pooled connections for the target
         :param endpoint: endpoint to target with the request
+        :param metadata: metadata to send with the request
         :param timeout: timeout for sending the requests
         :param retries: number of retries per gRPC call. If <0 it defaults to max(3, num_replicas)
         :return: list of asyncio.Task items for each send call
@@ -764,6 +770,7 @@ def send_request(
             shard_id=shard_id,
             polling_type=polling_type,
             endpoint=endpoint,
+            metadata=metadata,
             timeout=timeout,
             retries=retries,
         )
@@ -776,6 +783,7 @@ def send_requests(
         shard_id: Optional[int] = None,
         polling_type: PollingType = PollingType.ANY,
         endpoint: Optional[str] = None,
+        metadata: Optional[Dict[str, str]] = None,
         timeout: Optional[float] = None,
         retries: Optional[int] = -1,
     ) -> List[asyncio.Task]:
@@ -787,6 +795,7 @@ def send_requests(
         :param shard_id: Send to a specific shard of the deployment, ignored for polling ALL
         :param polling_type: defines if the request should be send to any or all pooled connections for the target
         :param endpoint: endpoint to target with the requests
+        :param metadata: metadata to send with the requests
         :param timeout: timeout for sending the requests
         :param retries: number of retries per gRPC call. If <0 it defaults to max(3, num_replicas)
         :return: list of asyncio.Task items for each send call
@@ -806,7 +815,12 @@ def send_requests(
 
         for replica_list in connections:
             task = self._send_requests(
-                requests, replica_list, endpoint, timeout=timeout, retries=retries
+                requests,
+                replica_list,
+                endpoint=endpoint,
+                metadata=metadata,
+                timeout=timeout,
+                retries=retries,
             )
             results.append(task)
 
@@ -846,6 +860,7 @@ def send_request_once(
         self,
         request: Request,
         deployment: str,
+        metadata: Optional[Dict[str, str]] = None,
         head: bool = False,
         shard_id: Optional[int] = None,
         timeout: Optional[float] = None,
@@ -854,6 +869,7 @@ def send_request_once(
         """Send msg to target via only one of the pooled connections
         :param request: request to send
         :param deployment: name of the Jina deployment to send the message to
+        :param metadata: metadata to send with the request
         :param head: If True it is send to the head, otherwise to the worker pods
         :param shard_id: Send to a specific shard of the deployment, ignored for polling ALL
         :param timeout: timeout for sending the requests
@@ -863,6 +879,7 @@ def send_request_once(
         return self.send_requests_once(
             [request],
             deployment=deployment,
+            metadata=metadata,
             head=head,
             shard_id=shard_id,
             timeout=timeout,
@@ -873,6 +890,7 @@ def send_requests_once(
         self,
         requests: List[Request],
         deployment: str,
+        metadata: Optional[Dict[str, str]] = None,
         head: bool = False,
         shard_id: Optional[int] = None,
         endpoint: Optional[str] = None,
@@ -883,6 +901,7 @@ def send_requests_once(
 
         :param requests: request to send
         :param deployment: name of the Jina deployment to send the request to
+        :param metadata: metadata to send with the request
         :param head: If True it is send to the head, otherwise to the worker pods
         :param shard_id: Send to a specific shard of the deployment, ignored for polling ALL
         :param endpoint: endpoint to target with the requests
@@ -893,7 +912,12 @@ def send_requests_once(
         replicas = self._connections.get_replicas(deployment, head, shard_id)
         if replicas:
             return self._send_requests(
-                requests, replicas, endpoint, timeout=timeout, retries=retries
+                requests,
+                replicas,
+                endpoint=endpoint,
+                metadata=metadata,
+                timeout=timeout,
+                retries=retries,
             )
         else:
             self._logger.debug(
@@ -1007,13 +1031,21 @@ def _send_requests(
         requests: List[Request],
         connections: ReplicaList,
         endpoint: Optional[str] = None,
+        metadata: Optional[Dict[str, str]] = None,
         timeout: Optional[float] = None,
         retries: Optional[int] = -1,
     ) -> asyncio.Task:
         # this wraps the awaitable object from grpc as a coroutine so it can be used as a task
         # the grpc call function is not a coroutine but some _AioCall
+
+        if endpoint:
+            metadata = metadata or {}
+            metadata['endpoint'] = endpoint
+
+        if metadata:
+            metadata = tuple(metadata.items())
+
         async def task_wrapper():
-            metadata = (('endpoint', endpoint),) if endpoint else None
             tried_addresses = set()
             if retries is None or retries < 0:
                 total_num_tries = (
diff --git a/jina/serve/runtimes/gateway/graph/topology_graph.py b/jina/serve/runtimes/gateway/graph/topology_graph.py
index 932ddee91b7a1..04082193623f4 100644
--- a/jina/serve/runtimes/gateway/graph/topology_graph.py
+++ b/jina/serve/runtimes/gateway/graph/topology_graph.py
@@ -35,6 +35,7 @@ def __init__(
             number_of_parts: int = 1,
             floating: bool = False,
             filter_condition: dict = None,
+            metadata: Optional[Dict] = None,
             reduce: bool = True,
             timeout_send: Optional[float] = None,
             retries: Optional[int] = -1,
@@ -48,6 +49,7 @@ def __init__(
             self.end_time = None
             self.status = None
             self._filter_condition = filter_condition
+            self._metadata = metadata
             self._reduce = reduce
             self._timeout_send = timeout_send
             self._retries = retries
@@ -59,12 +61,18 @@ def leaf(self):
 
         def _update_requests_with_filter_condition(self, need_copy):
             for i in range(len(self.parts_to_send)):
-                req = self.parts_to_send[i] if not need_copy else copy.deepcopy(self.parts_to_send[i])
+                req = (
+                    self.parts_to_send[i]
+                    if not need_copy
+                    else copy.deepcopy(self.parts_to_send[i])
+                )
                 filtered_docs = req.docs.find(self._filter_condition)
                 req.data.docs = filtered_docs
                 self.parts_to_send[i] = req
 
-        def _update_request_by_params(self, deployment_name: str, request_input_parameters: Dict):
+        def _update_request_by_params(
+            self, deployment_name: str, request_input_parameters: Dict
+        ):
             specific_parameters = _parse_specific_params(
                 request_input_parameters, deployment_name
             )
@@ -104,7 +112,7 @@ async def _wait_previous_and_send(
             executor_endpoint_mapping: Optional[Dict] = None,
             target_executor_pattern: Optional[str] = None,
             request_input_parameters: Dict = {},
-            copy_request_at_send: bool = False
+            copy_request_at_send: bool = False,
         ):
             # Check my condition and send request with the condition
             metadata = {}
@@ -126,7 +134,9 @@ async def _wait_previous_and_send(
                     self.start_time = datetime.utcnow()
                     self._update_request_by_params(self.name, request_input_parameters)
                     if self._filter_condition is not None:
-                        self._update_requests_with_filter_condition(need_copy=not copy_request_at_send)
+                        self._update_requests_with_filter_condition(
+                            need_copy=not copy_request_at_send
+                        )
                     if self._reduce and len(self.parts_to_send) > 1:
                         self.parts_to_send = [
                             DataRequestHandler.reduce_requests(self.parts_to_send)
@@ -150,6 +160,7 @@ async def _wait_previous_and_send(
                         resp, metadata = await connection_pool.send_requests_once(
                             requests=self.parts_to_send,
                             deployment=self.name,
+                            metadata=self._metadata,
                             head=True,
                             endpoint=endpoint,
                             timeout=self._timeout_send,
@@ -157,7 +168,9 @@ async def _wait_previous_and_send(
                         )
                         if DataRequestHandler._KEY_RESULT in resp.parameters:
                             # Accumulate results from each Node and then add them to the original
-                            self.result_in_params_returned = resp.parameters[DataRequestHandler._KEY_RESULT]
+                            self.result_in_params_returned = resp.parameters[
+                                DataRequestHandler._KEY_RESULT
+                            ]
                         request.parameters = request_input_parameters
                         resp.parameters = request_input_parameters
                         self.parts_to_send.clear()
@@ -181,7 +194,7 @@ def get_leaf_tasks(
             target_executor_pattern: Optional[str] = None,
             request_input_parameters: Dict = {},
             request_input_has_specific_params: bool = False,
-            copy_request_at_send: bool = False
+            copy_request_at_send: bool = False,
         ) -> List[Tuple[bool, asyncio.Task]]:
             """
             Gets all the tasks corresponding from all the subgraphs born from this node
@@ -227,7 +240,7 @@ def get_leaf_tasks(
                     executor_endpoint_mapping=executor_endpoint_mapping,
                     target_executor_pattern=target_executor_pattern,
                     request_input_parameters=request_input_parameters,
-                    copy_request_at_send=copy_request_at_send
+                    copy_request_at_send=copy_request_at_send,
                 )
             )
             if self.leaf:  # I am like a leaf
@@ -246,7 +259,8 @@ def get_leaf_tasks(
                     target_executor_pattern=target_executor_pattern,
                     request_input_parameters=request_input_parameters,
                     request_input_has_specific_params=request_input_has_specific_params,
-                    copy_request_at_send=num_outgoing_nodes > 1 and request_input_has_specific_params
+                    copy_request_at_send=num_outgoing_nodes > 1
+                    and request_input_has_specific_params,
                 )
                 # We are interested in the last one, that will be the task that awaits all the previous
                 hanging_tasks_tuples.extend(t)
@@ -307,6 +321,7 @@ def __init__(
         self,
         graph_representation: Dict,
         graph_conditions: Dict = {},
+        deployments_metadata: Dict = {},
         deployments_disable_reduce: List[str] = [],
         timeout_send: Optional[float] = 1.0,
         retries: Optional[int] = -1,
@@ -333,6 +348,7 @@ def __init__(
         nodes = {}
         for node_name in node_set:
             condition = graph_conditions.get(node_name, None)
+            metadata = deployments_metadata.get(node_name, None)
             nodes[node_name] = self._ReqReplyNode(
                 name=node_name,
                 number_of_parts=num_parts_per_node[node_name]
@@ -340,6 +356,7 @@ def __init__(
                 else 1,
                 floating=node_name in floating_deployment_set,
                 filter_condition=condition,
+                metadata=metadata,
                 reduce=node_name not in deployments_disable_reduce,
                 timeout_send=timeout_send,
                 retries=retries,
diff --git a/jina/serve/runtimes/gateway/grpc/gateway.py b/jina/serve/runtimes/gateway/grpc/gateway.py
index 6420736cb4016..c8c3a7e8ecf67 100644
--- a/jina/serve/runtimes/gateway/grpc/gateway.py
+++ b/jina/serve/runtimes/gateway/grpc/gateway.py
@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import TYPE_CHECKING, List, Optional
 
 import grpc
 from grpc_health.v1 import health, health_pb2, health_pb2_grpc
@@ -11,6 +11,9 @@
 from jina.serve.runtimes.helper import _get_grpc_server_options
 from jina.types.request.status import StatusMessage
 
+if TYPE_CHECKING:
+    from grpc.aio._interceptor import ServerInterceptor
+
 
 class GRPCGateway(BaseGateway):
     """GRPC Gateway implementation"""
@@ -37,13 +40,25 @@ def __init__(
         self.ssl_certfile = ssl_certfile
         self.health_servicer = health.HealthServicer(experimental_non_blocking=True)
 
+    def grpc_extra_server_interceptors(self) -> Optional[List['ServerInterceptor']]:
+        """Return a list of extra server interceptors to be added to the GRPC server.
+
+        :return: a list of extra server interceptors
+        """
+        return None
+
     async def setup_server(self):
         """
         setup GRPC server
         """
+        server_interceptors = self.grpc_tracing_server_interceptors or []
+        extra_interceptors = self.grpc_extra_server_interceptors()
+        if extra_interceptors:
+            server_interceptors.extend(extra_interceptors)
+
         self.server = grpc.aio.server(
             options=_get_grpc_server_options(self.grpc_server_options),
-            interceptors=self.grpc_tracing_server_interceptors,
+            interceptors=server_interceptors,
         )
 
         jina_pb2_grpc.add_JinaRPCServicer_to_server(
@@ -118,7 +133,6 @@ async def dry_run(self, empty, context) -> jina_pb2.StatusProto:
         :returns: the response request
         """
         from docarray import DocumentArray
-
         from jina.clients.request import request_generator
         from jina.enums import DataInputType
         from jina.serve.executors import __dry_run_endpoint__
diff --git a/jina/serve/streamer.py b/jina/serve/streamer.py
index cab1d7e3e080a..11e68f69eec61 100644
--- a/jina/serve/streamer.py
+++ b/jina/serve/streamer.py
@@ -1,7 +1,6 @@
 from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Union
 
 from docarray import DocumentArray
-
 from jina.logging.logger import JinaLogger
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.runtimes.gateway.graph.topology_graph import TopologyGraph
@@ -29,6 +28,7 @@ def __init__(
         graph_representation: Dict,
         executor_addresses: Dict[str, Union[str, List[str]]],
         graph_conditions: Dict = {},
+        deployments_metadata: Dict[str, Dict[str, str]] = {},
         deployments_disable_reduce: List[str] = [],
         timeout_send: Optional[float] = None,
         retries: int = 0,
@@ -47,6 +47,8 @@ def __init__(
             will be considered to be floating, and they will be "flagged" so that the user can ignore their tasks and not await them.
         :param executor_addresses: dictionary JSON with the input addresses of each Deployment. Each Executor can have one single address or a list of addrresses for each Executor
         :param graph_conditions: Dictionary stating which filtering conditions each Executor in the graph requires to receive Documents.
+        :param deployments_metadata: Dictionary with the metadata of each Deployment. Each executor deployment can have a list of key-value pairs to
+            provide information associated with the request to the deployment.
         :param deployments_disable_reduce: list of Executor disabling the built-in merging mechanism.
         :param timeout_send: Timeout to be considered when sending requests to Executors
         :param retries: Number of retries to try to make successfull sendings to Executors
@@ -62,6 +64,7 @@ def __init__(
         topology_graph = self._create_topology_graph(
             graph_representation,
             graph_conditions,
+            deployments_metadata,
             deployments_disable_reduce,
             timeout_send,
             retries,
@@ -95,6 +98,7 @@ def _create_topology_graph(
         self,
         graph_description,
         graph_conditions,
+        deployments_metadata,
         deployments_disable_reduce,
         timeout_send,
         retries,
@@ -103,6 +107,7 @@ def _create_topology_graph(
         return TopologyGraph(
             graph_representation=graph_description,
             graph_conditions=graph_conditions,
+            deployments_metadata=deployments_metadata,
             deployments_disable_reduce=deployments_disable_reduce,
             timeout_send=timeout_send,
             retries=retries,
diff --git a/jina_cli/autocomplete.py b/jina_cli/autocomplete.py
index 9a90c24598b33..34be3af89c32a 100644
--- a/jina_cli/autocomplete.py
+++ b/jina_cli/autocomplete.py
@@ -149,6 +149,7 @@
             '--graph-description',
             '--graph-conditions',
             '--deployments-addresses',
+            '--deployments-metadata',
             '--deployments-disable-reduce',
             '--compression',
             '--timeout-send',
@@ -212,8 +213,7 @@
             '--force',
         ],
         'hub status': ['--help', '--id', '--verbose', '--replay'],
-        'hub list': ['--help'],
-        'hub': ['--help', 'new', 'push', 'pull', 'status', 'list'],
+        'hub': ['--help', 'new', 'push', 'pull', 'status'],
         'cloud login': ['--help'],
         'cloud logout': ['--help'],
         'cloud deploy': ['--help', '--name', '--workspace', '--env-file'],
@@ -361,6 +361,7 @@
             '--uses-after',
             '--when',
             '--external',
+            '--grpc-metadata',
             '--deployment-role',
             '--tls',
         ],
diff --git a/tests/integration/clients_extra_kwargs/test_clients_post_extra_kwargs.py b/tests/integration/clients_extra_kwargs/test_clients_post_extra_kwargs.py
index 4b85211fb2e51..2c5888ada0bd9 100644
--- a/tests/integration/clients_extra_kwargs/test_clients_post_extra_kwargs.py
+++ b/tests/integration/clients_extra_kwargs/test_clients_post_extra_kwargs.py
@@ -39,12 +39,8 @@ async def intercept_service(self, continuation, handler_call_details):
             return self._deny
 
     class AlternativeGRPCGateway(GRPCGateway):
-        def __init__(self, *args, **kwargs):
-            super(AlternativeGRPCGateway, self).__init__(*args, **kwargs)
-            self.server = grpc.aio.server(
-                interceptors=(AuthInterceptor('access_key'),),
-                options=_get_grpc_server_options(self.grpc_server_options),
-            )
+        def grpc_extra_server_interceptors(self):
+            return [AuthInterceptor('access_key')]
 
     return Flow(protocol='grpc', uses=AlternativeGRPCGateway).add()
 
diff --git a/tests/integration/deployments/test_deployment.py b/tests/integration/deployments/test_deployment.py
index 946ec77190d9e..4e2ae5b768209 100644
--- a/tests/integration/deployments/test_deployment.py
+++ b/tests/integration/deployments/test_deployment.py
@@ -20,12 +20,14 @@ async def test_deployments_trivial_topology(port_generator):
     )
     deployments_addresses = f'{{"deployment0": ["0.0.0.0:{deployment_port}"]}}'
 
+    deployments_metadata = '{"deployment0": {"key": "value"}}'
+
     # create a single worker pod
     worker_deployment = _create_regular_deployment(deployment_port)
 
     # create a single gateway pod
     gateway_deployment = _create_gateway_deployment(
-        graph_description, deployments_addresses, port
+        graph_description, deployments_addresses, deployments_metadata, port
     )
 
     with gateway_deployment, worker_deployment:
@@ -93,10 +95,15 @@ async def test_deployments_flow_topology(
     deployments_addresses += '}'
     port = port_generator()
 
+    deployments_metadata = '{"deployment0": {"key": "value"}}'
+
     # create a single gateway pod
 
     gateway_deployment = _create_gateway_deployment(
-        json.dumps(complete_graph_dict), deployments_addresses, port
+        json.dumps(complete_graph_dict),
+        deployments_addresses,
+        deployments_metadata,
+        port,
     )
     gateway_deployment.start()
 
@@ -133,6 +140,7 @@ async def test_deployments_shards(polling, port_generator):
         '{"start-gateway": ["deployment0"], "deployment0": ["end-gateway"]}'
     )
     deployments_addresses = f'{{"deployment0": ["0.0.0.0:{head_port}"]}}'
+    deployments_metadata = '{"deployment0": {"key": "value"}}'
 
     deployment = _create_regular_deployment(
         port=head_port, name='deployment', polling=polling, shards=10
@@ -140,7 +148,7 @@ async def test_deployments_shards(polling, port_generator):
     deployment.start()
 
     gateway_deployment = _create_gateway_deployment(
-        graph_description, deployments_addresses, port
+        graph_description, deployments_addresses, deployments_metadata, port
     )
     gateway_deployment.start()
 
@@ -175,8 +183,9 @@ async def test_deployments_replicas(port_generator):
 
     connections = [f'0.0.0.0:{port}' for port in deployment.ports]
     deployments_addresses = f'{{"deployment0": {json.dumps(connections)}}}'
+    deployments_metadata = '{"deployment0": {"key": "value"}}'
     gateway_deployment = _create_gateway_deployment(
-        graph_description, deployments_addresses, port
+        graph_description, deployments_addresses, deployments_metadata, port
     )
     gateway_deployment.start()
 
@@ -203,6 +212,7 @@ async def test_deployments_with_executor(port_generator):
 
     head_port = port_generator()
     deployments_addresses = f'{{"deployment0": ["0.0.0.0:{head_port}"]}}'
+    deployments_metadata = '{"deployment0": {"key": "value"}}'
 
     regular_deployment = _create_regular_deployment(
         port=head_port,
@@ -217,7 +227,7 @@ async def test_deployments_with_executor(port_generator):
 
     port = port_generator()
     gateway_deployment = _create_gateway_deployment(
-        graph_description, deployments_addresses, port
+        graph_description, deployments_addresses, deployments_metadata, port
     )
     gateway_deployment.start()
 
@@ -256,9 +266,10 @@ async def test_deployments_with_replicas_advance_faster(port_generator):
 
     connections = [f'0.0.0.0:{port}' for port in deployment.ports]
     deployments_addresses = f'{{"deployment0": {json.dumps(connections)}}}'
+    deployments_metadata = '{"deployment0": {"key": "value"}}'
 
     gateway_deployment = _create_gateway_deployment(
-        graph_description, deployments_addresses, port
+        graph_description, deployments_addresses, deployments_metadata, port
     )
     gateway_deployment.start()
 
@@ -328,7 +339,9 @@ def _create_regular_deployment(
     return Deployment(args)
 
 
-def _create_gateway_deployment(graph_description, deployments_addresses, port):
+def _create_gateway_deployment(
+    graph_description, deployments_addresses, deployments_metadata, port
+):
     return Deployment(
         set_gateway_parser().parse_args(
             [
@@ -336,6 +349,8 @@ def _create_gateway_deployment(graph_description, deployments_addresses, port):
                 graph_description,
                 '--deployments-addresses',
                 deployments_addresses,
+                '--deployments-metadata',
+                deployments_metadata,
                 '--port',
                 str(port),
             ]
diff --git a/tests/integration/external_deployment/test_external_deployment.py b/tests/integration/external_deployment/test_external_deployment.py
index 39dea6effeff8..93edf3f6ffb6c 100644
--- a/tests/integration/external_deployment/test_external_deployment.py
+++ b/tests/integration/external_deployment/test_external_deployment.py
@@ -2,6 +2,7 @@
 
 import pytest
 
+import jina.excepts
 from jina import Document, DocumentArray, Executor, Flow, requests
 from jina.helper import random_port
 from jina.orchestrate.deployments import Deployment
@@ -347,3 +348,68 @@ def foo(self, docs, **kwargs):
             response = f.post(on='/', inputs=d, target_executor='external_executor')
 
     assert response[0].text == 'external'
+
+
+def test_external_flow_with_grpc_metadata():
+    import grpc
+
+    from jina.serve.runtimes.gateway import GRPCGateway
+
+    class DummyInterceptor(grpc.aio.ServerInterceptor):
+        def __init__(self):
+            def deny(_, context):
+                context.abort(grpc.StatusCode.UNAUTHENTICATED, 'Invalid key')
+
+            self._deny = grpc.unary_unary_rpc_method_handler(deny)
+
+        async def intercept_service(self, continuation, handler_call_details):
+            meta = handler_call_details.invocation_metadata
+            method = handler_call_details.method
+            if method != '/jina.JinaRPC/Call':
+                return await continuation(handler_call_details)
+
+            for m in meta:
+                if m == ('authorization', 'valid'):
+                    return await continuation(handler_call_details)
+
+            return self._deny
+
+    class DummyGRPCGateway(GRPCGateway):
+        def __int__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+
+        def grpc_extra_server_interceptors(self):
+            return [DummyInterceptor()]
+
+    class ExtExecutor(Executor):
+        @requests
+        def foo(self, docs, **kwargs):
+            for doc in docs:
+                doc.text = 'external'
+
+    external_flow = Flow(uses=DummyGRPCGateway).add(uses=ExtExecutor)
+
+    with external_flow:
+        d = Document(text='sunset with green landscape by the river')
+        f = Flow().add(
+            port=external_flow.port,
+            external=True,
+            name='external_executor',
+            grpc_metadata={'authorization': 'valid'},
+        )
+        with f:
+            response = f.post(on='/', inputs=d, target_executor='external_executor')
+
+            assert response[0].text == 'external'
+
+        f = Flow().add(
+            port=external_flow.port,
+            external=True,
+            name='external_executor',
+            grpc_metadata={'authorization': 'invalid'},
+        )
+        with f:
+            with pytest.raises(jina.excepts.BadServerFlow) as error_info:
+                response = f.post(on='/', inputs=d, target_executor='external_executor')
+            assert 'Invalid key' in str(error_info.value)
+            # assert response[0].text == 'external'
diff --git a/tests/integration/gateway_clients/test_clients_gateways.py b/tests/integration/gateway_clients/test_clients_gateways.py
index d4ea3d41729df..f8e11d71c9ccc 100644
--- a/tests/integration/gateway_clients/test_clients_gateways.py
+++ b/tests/integration/gateway_clients/test_clients_gateways.py
@@ -84,6 +84,7 @@ def send_requests_once(
         requests,
         deployment: str,
         head: bool,
+        metadata: dict = None,
         endpoint: str = None,
         timeout: float = 1.0,
         retries: int = -1,
@@ -109,6 +110,7 @@ def send_requests_once(
         requests,
         deployment: str,
         head: bool,
+        metadata: dict = None,
         shard_id=None,
         endpoint: str = None,
         timeout: float = 1.0,
diff --git a/tests/integration/gateway_clients/test_streaming.py b/tests/integration/gateway_clients/test_streaming.py
index 8cd3190b0d5ed..000bcb9b341c4 100644
--- a/tests/integration/gateway_clients/test_streaming.py
+++ b/tests/integration/gateway_clients/test_streaming.py
@@ -62,6 +62,7 @@ def send_requests_once(
         requests,
         deployment: str,
         head: bool,
+        metadata: dict = None,
         shard_id=None,
         endpoint: str = None,
         timeout: float = 1.0,
diff --git a/tests/k8s/test_k8s.py b/tests/k8s/test_k8s.py
index 337ea5e12b7ef..f1916240109d4 100644
--- a/tests/k8s/test_k8s.py
+++ b/tests/k8s/test_k8s.py
@@ -6,9 +6,9 @@
 import pytest
 import requests as req
 import yaml
-from docarray import DocumentArray
 from pytest_kind import cluster
 
+from docarray import DocumentArray
 from jina import Document, Executor, Flow, requests
 from jina.orchestrate.deployments import Deployment
 from jina.orchestrate.deployments.config.k8s import K8sDeploymentConfig
@@ -666,6 +666,104 @@ async def test_flow_with_external_k8s_deployment(logger, docker_images, tmpdir):
         assert 'workspace' in doc.tags
 
 
+@pytest.mark.asyncio
+@pytest.mark.timeout(3600)
+@pytest.mark.parametrize('grpc_metadata', [{}, {"key1": "value1"}])
+async def test_flow_with_metadata_k8s_deployment(logger, grpc_metadata, tmpdir):
+    docker_images = ['test-executor', 'jinaai/jina']
+
+    namespace = 'test-flow-with-metadata-k8s-deployment'
+    from kubernetes import client
+
+    api_client = client.ApiClient()
+    core_client = client.CoreV1Api(api_client=api_client)
+    app_client = client.AppsV1Api(api_client=api_client)
+
+    await _create_external_deployment(api_client, app_client, docker_images, tmpdir)
+
+    flow = Flow(name='k8s_flow-with_metadata_deployment', port=9090).add(
+        name='external_executor',
+        external=True,
+        host='external-deployment.external-deployment-ns.svc',
+        port=GrpcConnectionPool.K8S_PORT,
+        grpc_metadata=grpc_metadata,
+    )
+
+    dump_path = os.path.join(str(tmpdir), namespace)
+    flow.to_kubernetes_yaml(dump_path, k8s_namespace=namespace)
+
+    await create_all_flow_deployments_and_wait_ready(
+        dump_path,
+        namespace=namespace,
+        api_client=api_client,
+        app_client=app_client,
+        core_client=core_client,
+        deployment_replicas_expected={
+            'gateway': 1,
+        },
+        logger=logger,
+    )
+
+    resp = await run_test(
+        flow=flow,
+        namespace=namespace,
+        core_client=core_client,
+        endpoint='/workspace',
+    )
+    docs = resp[0].docs
+    for doc in docs:
+        assert 'workspace' in doc.tags
+
+
+@pytest.mark.asyncio
+@pytest.mark.timeout(3600)
+@pytest.mark.parametrize('grpc_metadata', [{}, {"key1": "value1"}])
+async def test_flow_with_metadata_k8s_deployment(logger, grpc_metadata, tmpdir):
+    docker_images = ['test-executor', 'jinaai/jina']
+
+    namespace = 'test-flow-with-metadata-k8s-deployment'
+    from kubernetes import client
+
+    api_client = client.ApiClient()
+    core_client = client.CoreV1Api(api_client=api_client)
+    app_client = client.AppsV1Api(api_client=api_client)
+
+    await _create_external_deployment(api_client, app_client, docker_images, tmpdir)
+
+    flow = Flow(name='k8s_flow-with_metadata_deployment', port=9090).add(
+        name='external_executor',
+        external=True,
+        host='external-deployment.external-deployment-ns.svc',
+        port=GrpcConnectionPool.K8S_PORT,
+        grpc_metadata=grpc_metadata,
+    )
+
+    dump_path = os.path.join(str(tmpdir), namespace)
+    flow.to_kubernetes_yaml(dump_path, k8s_namespace=namespace)
+
+    await create_all_flow_deployments_and_wait_ready(
+        dump_path,
+        namespace=namespace,
+        api_client=api_client,
+        app_client=app_client,
+        core_client=core_client,
+        deployment_replicas_expected={
+            'gateway': 1,
+        },
+        logger=logger,
+    )
+
+    resp = await run_test(
+        flow=flow,
+        namespace=namespace,
+        core_client=core_client,
+        endpoint='/workspace',
+    )
+    docs = resp[0].docs
+    for doc in docs:
+        assert 'workspace' in doc.tags
+
+
 async def _create_external_deployment(api_client, app_client, docker_images, tmpdir):
     namespace = 'external-deployment-ns'
     args = set_deployment_parser().parse_args(
diff --git a/tests/unit/orchestrate/deployments/test_deployments.py b/tests/unit/orchestrate/deployments/test_deployments.py
index af8cd43568cc8..050bcf0261df8 100644
--- a/tests/unit/orchestrate/deployments/test_deployments.py
+++ b/tests/unit/orchestrate/deployments/test_deployments.py
@@ -148,6 +148,16 @@ def test_pod_context_shards_replicas(shards):
     Deployment(args).start().close()
 
 
+@pytest.mark.parametrize('metadata', [{'key1': 'value1', 'key2': 'value2'}])
+def test_pod_context_grpc_metadata(metadata):
+    args_list = []
+    for k, v in metadata.items():
+        args_list.extend(['--grpc-metadata', f'{k}:{v}'])
+    args = set_deployment_parser().parse_args(args_list)
+    with Deployment(args) as bp:
+        assert bp.grpc_metadata == metadata
+
+
 class AppendNameExecutor(Executor):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
diff --git a/tests/unit/orchestrate/flow/flow-construct/test_flow.py b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
index 35db758e66af3..716e5d4e1c6ad 100644
--- a/tests/unit/orchestrate/flow/flow-construct/test_flow.py
+++ b/tests/unit/orchestrate/flow/flow-construct/test_flow.py
@@ -617,3 +617,24 @@ def test_set_port_deployment(port_generator):
     with Flow().add(uses=Executor, port=port) as f:
         assert int(f._deployment_nodes['executor0'].pod_args['pods'][0][0].port) == port
         f.index(inputs=[])
+
+
+def test_set_deployment_grpc_metadata():
+    f = Flow().add(name='my_exec', uses=Executor, grpc_metadata={'key': 'value'})
+
+    with f:
+        metadata = f._get_deployments_metadata()
+        assert metadata['my_exec'] == {'key': 'value'}
+
+        k8s_metadata = f._get_k8s_deployments_metadata()
+        assert metadata == k8s_metadata
+
+        assert f._deployment_nodes['gateway'].args.deployments_metadata == json.dumps(
+            metadata
+        )
+
+        assert f._deployment_nodes['my_exec'].pod_args['pods'][0][0].grpc_metadata == {
+            'key': 'value'
+        }
+
+        f.post('/', inputs=Document())
diff --git a/tests/unit/serve/gateway/test_gateway.py b/tests/unit/serve/gateway/test_gateway.py
index a8f64bb3cd4e5..581f87125686b 100644
--- a/tests/unit/serve/gateway/test_gateway.py
+++ b/tests/unit/serve/gateway/test_gateway.py
@@ -4,8 +4,8 @@
 import time
 
 import pytest
-from docarray import DocumentArray
 
+from docarray import DocumentArray
 from jina import Executor, requests
 from jina.helper import random_port
 from jina.parsers import set_gateway_parser, set_pod_parser
@@ -32,6 +32,7 @@ def process(self, docs: DocumentArray, **kwargs):
 def _create_gateway_runtime(port, uses, uses_with, worker_port):
     graph_description = '{"start-gateway": ["pod0"], "pod0": ["end-gateway"]}'
     pod_addresses = f'{{"pod0": ["0.0.0.0:{worker_port}"]}}'
+    deployments_metadata = '{"pod0": {"key1": "value1", "key2": "value2"}}'
     with GatewayRuntime(
         set_gateway_parser().parse_args(
             [
@@ -45,6 +46,8 @@ def _create_gateway_runtime(port, uses, uses_with, worker_port):
                 graph_description,
                 '--deployments-addresses',
                 pod_addresses,
+                '--deployments-metadata',
+                deployments_metadata,
             ]
         )
     ) as runtime:
diff --git a/tests/unit/serve/runtimes/gateway/graph/test_topology_graph.py b/tests/unit/serve/runtimes/gateway/graph/test_topology_graph.py
index 1abe3b7bce4e0..9e55d73631a79 100644
--- a/tests/unit/serve/runtimes/gateway/graph/test_topology_graph.py
+++ b/tests/unit/serve/runtimes/gateway/graph/test_topology_graph.py
@@ -551,6 +551,7 @@ def send_requests_once(
         requests: List[Request],
         deployment: str,
         head: bool,
+        metadata: dict = None,
         endpoint: str = None,
         timeout: float = 1.0,
         retries: int = -1,
@@ -580,10 +581,21 @@ async def task_wrapper():
 
 
 class DummyMockGatewayRuntime:
-    def __init__(self, graph_representation, conditions={}, *args, **kwargs):
+    def __init__(
+        self,
+        graph_representation,
+        conditions={},
+        deployments_metadata={},
+        *args,
+        **kwargs,
+    ):
         super().__init__(*args, **kwargs)
         self.connection_pool = DummyMockConnectionPool(*args, **kwargs)
-        self.graph = TopologyGraph(graph_representation, conditions)
+        self.graph = TopologyGraph(
+            graph_representation,
+            graph_conditions=conditions,
+            deployments_metadata=deployments_metadata,
+        )
 
     async def receive_from_client(self, client_id, msg: 'DataRequest'):
         graph = copy.deepcopy(self.graph)
@@ -966,6 +978,48 @@ async def test_message_ordering_two_joins_graph(
         assert path02 or path03 or path12 or path13
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'deployments_metadata',
+    [
+        ({}),
+        (
+            {
+                'deployment1': {'key1': 'value1'},
+                'deployment2': {'key2': 'value2'},
+            }
+        ),
+    ],
+)
+async def test_deployment_metadata_in_graph(linear_graph_dict, deployments_metadata):
+    runtime = DummyMockGatewayRuntime(
+        linear_graph_dict, deployments_metadata=deployments_metadata
+    )
+    for node in runtime.graph.origin_nodes:
+        if node.name in deployments_metadata:
+            assert node._metadata == deployments_metadata[node.name]
+
+    resps = await asyncio.gather(
+        runtime.receive_from_client(0, create_req_from_text('client0-Request')),
+        runtime.receive_from_client(1, create_req_from_text('client1-Request')),
+        runtime.receive_from_client(2, create_req_from_text('client2-Request')),
+        runtime.receive_from_client(3, create_req_from_text('client3-Request')),
+        runtime.receive_from_client(4, create_req_from_text('client4-Request')),
+        runtime.receive_from_client(5, create_req_from_text('client5-Request')),
+        runtime.receive_from_client(6, create_req_from_text('client6-Request')),
+        runtime.receive_from_client(7, create_req_from_text('client7-Request')),
+        runtime.receive_from_client(8, create_req_from_text('client8-Request')),
+        runtime.receive_from_client(9, create_req_from_text('client9-Request')),
+    )
+    assert len(resps) == 10
+    for client_id, client_resps in resps:
+        assert len(client_resps) == 1
+        assert (
+            f'client{client_id}-Request-client{client_id}-deployment0-client{client_id}-deployment1-client{client_id}-deployment2-client{client_id}-deployment3'
+            == client_resps[0].docs[0].text
+        )
+
+
 def test_empty_graph():
     graph = TopologyGraph({})
     assert not graph.origin_nodes
diff --git a/tests/unit/serve/runtimes/gateway/grpc/test_grpc_gateway_runtime.py b/tests/unit/serve/runtimes/gateway/grpc/test_grpc_gateway_runtime.py
index a0686c5b2df5d..e4e69228767f9 100644
--- a/tests/unit/serve/runtimes/gateway/grpc/test_grpc_gateway_runtime.py
+++ b/tests/unit/serve/runtimes/gateway/grpc/test_grpc_gateway_runtime.py
@@ -122,6 +122,7 @@ def send_requests_once(
         requests,
         deployment: str,
         head: bool,
+        metadata: dict = None,
         shard_id=None,
         endpoint: str = None,
         timeout: float = 1.0,
diff --git a/tests/unit/serve/runtimes/head/test_head_runtime.py b/tests/unit/serve/runtimes/head/test_head_runtime.py
index 92226b8812611..4aafe24ccca2b 100644
--- a/tests/unit/serve/runtimes/head/test_head_runtime.py
+++ b/tests/unit/serve/runtimes/head/test_head_runtime.py
@@ -275,6 +275,7 @@ def _send_requests_mock(
             request: List[Request],
             connection,
             endpoint,
+            metadata: dict = None,
             timeout=1.0,
             retries=-1,
         ) -> asyncio.Task:


Code review comments:
diff_hunk: @@ -37,13 +40,25 @@ def __init__(
         self.ssl_certfile = ssl_certfile
         self.health_servicer = health.HealthServicer(experimental_non_blocking=True)
 
+    def grpc_extra_server_interceptors(self) -> Optional[List['ServerInterceptor']]:
path: jina/serve/runtimes/gateway/grpc/gateway.py
text: @user1:
This decision will be made in the scope of the CustomGateway design, we will decide which methods are offered and which not. For now, user needs to override `setup_server` so please do so.