jina.serve.networking module#

class jina.serve.networking.ReplicaList[source]#

Bases: object

Maintains a list of connections to replicas and uses round robin for selecting a replica

add_connection(address)[source]#

Add connection with address to the connection list :type address: str :param address: Target address of this connection

async remove_connection(address)[source]#

Remove connection with address from the connection list :type address: str :param address: Remove connection for this address :returns: The removed connection or None if there was not any for the given address

get_next_connection()[source]#

Returns a connection from the list. Strategy is round robin :returns: A connection from the pool

get_all_connections()[source]#

Returns all available connections :returns: A complete list of all connections from the pool

has_connection(address)[source]#

Checks if a connection for ip exists in the list :type address: str :param address: The address to check :rtype: bool :returns: True if a connection for the ip exists in the list

has_connections()[source]#

Checks if this contains any connection :rtype: bool :returns: True if any connection is managed, False otherwise

async close()[source]#

Close all connections and clean up internal state

class jina.serve.networking.GrpcConnectionPool(logger=None)[source]#

Bases: object

Manages a list of grpc connections.

Parameters

logger (Optional[JinaLogger]) – the logger to use

send_request(request, deployment, head=False, shard_id=None, polling_type=PollingType.ANY, endpoint=None)[source]#

Send a single message to target via one or all of the pooled connections, depending on polling_type. Convenience function wrapper around send_request. :type request: Request :param request: a single request to send :type deployment: str :param deployment: name of the Jina deployment to send the message to :type head: bool :param head: If True it is send to the head, otherwise to the worker pods :type shard_id: Optional[int] :param shard_id: Send to a specific shard of the deployment, ignored for polling ALL :type polling_type: PollingType :param polling_type: defines if the message should be send to any or all pooled connections for the target :type endpoint: Optional[str] :param endpoint: endpoint to target with the request :rtype: List[Task] :return: list of asyncio.Task items for each send call

send_requests(requests, deployment, head=False, shard_id=None, polling_type=PollingType.ANY, endpoint=None)[source]#

Send a request to target via one or all of the pooled connections, depending on polling_type

Parameters

  • requests (List[Request]) – request (DataRequest/ControlRequest) to send

  • deployment (str) – name of the Jina deployment to send the request to

  • head (bool) – If True it is send to the head, otherwise to the worker pods

  • shard_id (Optional[int]) – Send to a specific shard of the deployment, ignored for polling ALL

  • polling_type (PollingType) – defines if the request should be send to any or all pooled connections for the target

  • endpoint (Optional[str]) – endpoint to target with the requests

Return type

List[Task]

Returns

list of asyncio.Task items for each send call

send_request_once(request, deployment, head=False, shard_id=None)[source]#

Send msg to target via only one of the pooled connections :type request: Request :param request: request to send :type deployment: str :param deployment: name of the Jina deployment to send the message to :type head: bool :param head: If True it is send to the head, otherwise to the worker pods :type shard_id: Optional[int] :param shard_id: Send to a specific shard of the deployment, ignored for polling ALL :rtype: Task :return: asyncio.Task representing the send call

send_requests_once(requests, deployment, head=False, shard_id=None, endpoint=None)[source]#

Send a request to target via only one of the pooled connections

Parameters

  • requests (List[Request]) – request to send

  • deployment (str) – name of the Jina deployment to send the request to

  • head (bool) – If True it is send to the head, otherwise to the worker pods

  • shard_id (Optional[int]) – Send to a specific shard of the deployment, ignored for polling ALL

  • endpoint (Optional[str]) – endpoint to target with the requests

Return type

Task

Returns

asyncio.Task representing the send call

add_connection(deployment, address, head=False, shard_id=None)[source]#

Adds a connection for a deployment to this connection pool

Parameters

  • deployment (str) – The deployment the connection belongs to, like ‘encoder’

  • head (Optional[bool]) – True if the connection is for a head

  • address (str) – Address used for the grpc connection, format is <host>:<port>

  • shard_id (Optional[int]) – Optional parameter to indicate this connection belongs to a shard, ignored for heads

async remove_connection(deployment, address, head=False, shard_id=None)[source]#

Removes a connection to a deployment

Parameters

  • deployment (str) – The deployment the connection belongs to, like ‘encoder’

  • address (str) – Address used for the grpc connection, format is <host>:<port>

  • head (Optional[bool]) – True if the connection is for a head

  • shard_id (Optional[int]) – Optional parameter to indicate this connection belongs to a shard, ignored for heads

Returns

The removed connection, None if it did not exist

start()[source]#

Starts the connection pool

async close()[source]#

Closes the connection pool

static get_grpc_channel(address, options=None, asyncio=False, https=False, root_certificates=None)[source]#

Creates a grpc channel to the given address

Parameters

  • address (str) – The address to connect to, format is <host>:<port>

  • options (Optional[list]) – A list of options to pass to the grpc channel

  • asyncio (Optional[bool]) – If True, use the asyncio implementation of the grpc channel

  • https (Optional[bool]) – If True, use https for the grpc channel

  • root_certificates (Optional[str]) – The path to the root certificates for https, only used if https is True

Return type

Channel

Returns

A grpc channel or an asyncio channel

static activate_worker_sync(worker_host, worker_port, target_head, shard_id=None)[source]#

Register a given worker to a head by sending an activate request

Parameters

  • worker_host (str) – the host address of the worker

  • worker_port (int) – the port of the worker

  • target_head (str) – address of the head to send the activate request to

  • shard_id (Optional[int]) – id of the shard the worker belongs to

Return type

ControlRequest

Returns

the response request

async static activate_worker(worker_host, worker_port, target_head, shard_id=None)[source]#

Register a given worker to a head by sending an activate request

Parameters

  • worker_host (str) – the host address of the worker

  • worker_port (int) – the port of the worker

  • target_head (str) – address of the head to send the activate request to

  • shard_id (Optional[int]) – id of the shard the worker belongs to

Return type

ControlRequest

Returns

the response request

async static deactivate_worker(worker_host, worker_port, target_head, shard_id=None)[source]#

Remove a given worker to a head by sending a deactivate request

Parameters

  • worker_host (str) – the host address of the worker

  • worker_port (int) – the port of the worker

  • target_head (str) – address of the head to send the deactivate request to

  • shard_id (Optional[int]) – id of the shard the worker belongs to

Return type

ControlRequest

Returns

the response request

static send_request_sync(request, target, timeout=100.0, https=False, root_certificates=None, endpoint=None)[source]#

Sends a request synchronically to the target via grpc

Parameters

  • request (Request) – the request to send

  • target (str) – where to send the request to, like 127.0.0.1:8080

  • timeout – timeout for the send

  • https – if True, use https for the grpc channel

  • root_certificates (Optional[str]) – the path to the root certificates for https, only used if https is True

  • endpoint (Optional[str]) – endpoint to target with the request

Return type

Request

Returns

the response request

static send_requests_sync(requests, target, timeout=100.0, https=False, root_certificates=None, endpoint=None)[source]#

Sends a list of requests synchronically to the target via grpc

Parameters

  • requests (List[Request]) – the requests to send

  • target (str) – where to send the request to, like 127.0.0.1:8080

  • timeout – timeout for the send

  • https – if True, use https for the grpc channel

  • root_certificates (Optional[str]) – the path to the root certificates for https, only used if https is True

  • endpoint (Optional[str]) – endpoint to target with the request

Return type

Request

Returns

the response request

static get_default_grpc_options()[source]#

Returns a list of default options used for creating grpc channels. Documentation is here https://github.com/grpc/grpc/blob/master/include/grpc/impl/codegen/grpc_types.h :returns: list of tuples defining grpc parameters

async static send_request_async(request, target, timeout=1.0, https=False, root_certificates=None)[source]#

Sends a request asynchronously to the target via grpc

Parameters

  • request (Request) – the request to send

  • target (str) – where to send the request to, like 127.0.0.1:8080

  • timeout (float) – timeout for the send

  • https (bool) – if True, use https for the grpc channel

  • root_certificates (Optional[str]) – the path to the root certificates for https, only u

Return type

Request

Returns

the response request

static create_async_channel_stub(address, https=False, root_certificates=None)[source]#

Creates an async GRPC Channel. This channel has to be closed eventually!

Parameters

  • address – the address to create the connection to, like 127.0.0.0.1:8080

  • https – if True, use https for the grpc channel

  • root_certificates (Optional[str]) – the path to the root certificates for https, only u

Return type

Tuple[JinaSingleDataRequestRPCStub, JinaDataRequestRPCStub, JinaControlRequestRPCStub, Channel]

Returns

DataRequest/ControlRequest stubs and an async grpc channel

class jina.serve.networking.K8sGrpcConnectionPool(namespace, client, logger=None)[source]#

Bases: GrpcConnectionPool

Manages grpc connections to replicas in a K8s deployment.

Parameters

  • namespace (str) – K8s namespace to operate in

  • client (kubernetes.client.CoreV1Api) – K8s client

  • logger (JinaLogger) – the logger to use

K8S_PORT = 8080#
K8S_PORT_USES_BEFORE = 8081#
K8S_PORT_USES_AFTER = 8082#
start()[source]#

Subscribe to the K8s API and watch for changes in Pods

run()[source]#

Subscribes on MODIFIED events from list_namespaced_pod AK8s PI

async close()[source]#

Closes the connection pool

jina.serve.networking.is_remote_local_connection(first, second)[source]#

Decides, whether first is remote host and second is localhost

Parameters

  • first (str) – the ip or host name of the first runtime

  • second (str) – the ip or host name of the second runtime

Returns

True, if first is remote and second is local

jina.serve.networking.create_connection_pool(k8s_connection_pool=False, k8s_namespace=None, logger=None)[source]#

Creates the appropriate connection pool based on parameters :type k8s_namespace: Optional[str] :param k8s_namespace: k8s namespace the pool will live in, None if outside K8s :type k8s_connection_pool: bool :param k8s_connection_pool: flag to indicate if K8sGrpcConnectionPool should be used, defaults to true in K8s :type logger: Optional[JinaLogger] :param logger: the logger to use :rtype: GrpcConnectionPool :return: A connection pool object

jina.serve.networking.host_is_local(hostname)[source]#

Check if hostname is point to localhost :param hostname: host to check :return: True if hostname means localhost, False otherwise