function.runtime

  1# Copyright 2023 The Crossplane Authors.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     http://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""Utilities to create a composition function runtime."""
 16
 17import asyncio
 18import os
 19import signal
 20from collections.abc import Sequence
 21from typing import Any
 22
 23import grpc
 24from grpc_reflection.v1alpha import reflection
 25
 26import crossplane.function.proto.v1.run_function_pb2 as fnv1
 27import crossplane.function.proto.v1.run_function_pb2_grpc as grpcv1
 28import crossplane.function.proto.v1beta1.run_function_pb2 as fnv1beta1
 29import crossplane.function.proto.v1beta1.run_function_pb2_grpc as grpcv1beta1
 30
 31SERVICE_NAMES = (
 32    reflection.SERVICE_NAME,
 33    fnv1.DESCRIPTOR.services_by_name["FunctionRunnerService"].full_name,
 34    fnv1beta1.DESCRIPTOR.services_by_name["FunctionRunnerService"].full_name,
 35)
 36
 37SHUTDOWN_GRACE_PERIOD_SECONDS = 5
 38
 39
 40def load_credentials(tls_certs_dir: str) -> grpc.ServerCredentials:
 41    """Load TLS credentials for a composition function gRPC server.
 42
 43    Args:
 44        tls_certs_dir: A directory containing tls.crt, tls.key, and ca.crt.
 45
 46    Returns:
 47        gRPC mTLS server credentials.
 48
 49    tls.crt and tls.key must be the function's PEM-encoded certificate and
 50    private key. ca.cert must be a PEM-encoded CA certificate used to
 51    authenticate callers (i.e. Crossplane).
 52    """
 53    if tls_certs_dir is None:
 54        return None
 55
 56    with open(os.path.join(tls_certs_dir, "tls.crt"), "rb") as f:
 57        crt = f.read()
 58
 59    with open(os.path.join(tls_certs_dir, "tls.key"), "rb") as f:
 60        key = f.read()
 61
 62    with open(os.path.join(tls_certs_dir, "ca.crt"), "rb") as f:
 63        ca = f.read()
 64
 65    return grpc.ssl_server_credentials(
 66        private_key_certificate_chain_pairs=[(key, crt)],
 67        root_certificates=ca,
 68        require_client_auth=True,
 69    )
 70
 71
 72def serve(
 73    function: grpcv1.FunctionRunnerServiceServicer,
 74    address: str,
 75    *,
 76    creds: grpc.ServerCredentials,
 77    insecure: bool,
 78    options: Sequence[tuple[str, Any]] | None = None,
 79) -> None:
 80    """Start a gRPC server and serve requests asychronously.
 81
 82    Args:
 83        function: The function (class) to use to serve requests.
 84        address: The address at which to listen for requests.
 85        creds: The credentials used to authenticate requests.
 86        insecure: Serve insecurely, without credentials or encryption.
 87        options: Additional gRPC server options. Eg. set max receive message
 88                 size to 5Mb (default is 4Mb):
 89                 [("grpc.max_receive_message_length", 1024 * 1024 * 5)]
 90
 91    Raises:
 92        ValueError if creds is None and insecure is False.
 93
 94    If insecure is true requests will be served insecurely, even if credentials
 95    are supplied.
 96    """
 97    # Define the loop before the server so everything uses the same loop.
 98    try:
 99        loop = asyncio.get_event_loop()
100    except RuntimeError:
101        loop = asyncio.new_event_loop()
102        asyncio.set_event_loop(loop)
103
104    server = grpc.aio.server(options=options)
105
106    loop.add_signal_handler(
107        signal.SIGTERM,
108        lambda: asyncio.ensure_future(server.stop(grace=SHUTDOWN_GRACE_PERIOD_SECONDS)),
109    )
110
111    grpcv1.add_FunctionRunnerServiceServicer_to_server(function, server)
112    grpcv1beta1.add_FunctionRunnerServiceServicer_to_server(
113        BetaFunctionRunner(wrapped=function), server
114    )
115    reflection.enable_server_reflection(SERVICE_NAMES, server)
116
117    if creds is None and insecure is False:
118        msg = (
119            "no credentials were provided - did you provide credentials or use "
120            "the insecure flag?"
121        )
122        raise ValueError(msg)
123
124    if creds is not None:
125        server.add_secure_port(address, creds)
126
127    if insecure:
128        server.add_insecure_port(address)
129
130    async def start():
131        await server.start()
132        await server.wait_for_termination()
133
134    try:
135        loop.run_until_complete(start())
136    finally:
137        loop.run_until_complete(server.stop(grace=SHUTDOWN_GRACE_PERIOD_SECONDS))
138        loop.close()
139
140
141class BetaFunctionRunner(grpcv1beta1.FunctionRunnerServiceServicer):
142    """A BetaFunctionRunner handles beta gRPC RunFunctionRequests.
143
144    It handles requests by passing them to a wrapped v1.FunctionRunnerServiceServicer.
145    Incoming v1beta1 requests are converted to v1 by round-tripping them through
146    serialization. Outgoing requests are converted from v1 to v1beta1 the same
147    way.
148    """
149
150    def __init__(self, wrapped: grpcv1.FunctionRunnerServiceServicer):
151        """Create a new BetaFunctionRunner."""
152        self.wrapped = wrapped
153
154    async def RunFunction(  # noqa: N802  # gRPC requires this name. # pyright: ignore[reportIncompatibleMethodOverride]
155        self, req: fnv1beta1.RunFunctionRequest, context: grpc.aio.ServicerContext
156    ) -> fnv1beta1.RunFunctionResponse:
157        """Run the underlying function."""
158        gareq = fnv1.RunFunctionRequest()
159        gareq.ParseFromString(req.SerializeToString())
160
161        garsp = await self.wrapped.RunFunction(gareq, context)
162        rsp = fnv1beta1.RunFunctionResponse()
163        rsp.ParseFromString(garsp.SerializeToString())
164
165        return rsp