Implement ConfigMap and Secret based python packages.

Author: Patrick J. McNerthney

Dockerfile

@@ -1,7 +1,7 @@
 # syntax=docker/dockerfile:1
 
-# It's important that this is Debian 12 to match the distroless image.
-FROM debian:12-slim AS build
+# It's important that this is Debian to match the python image.
+FROM debian:trixie-slim AS build
 
 #RUN --mount=type=cache,target=/var/lib/apt/lists \
 #    --mount=type=cache,target=/var/cache/apt \
@@ -35,9 +35,13 @@ RUN \
 
 # Copy the pythonic venv to our runtime stage. It's important that the path be
 # the same as in the build stage, to avoid shebang paths and symlinks breaking. 
-FROM gcr.io/distroless/python3-debian12 AS image
-WORKDIR /
-USER nonroot:nonroot
-COPY --from=build --chown=nonroot:nonroot /venv/fn /venv/fn
+FROM python:3.13-slim-trixie AS image
+RUN \
+  addgroup --gid 2000 pythonic && \
+  adduser --uid 2000 --ingroup pythonic --disabled-password --no-create-home --disabled-login pythonic
+USER pythonic:pythonic
+COPY --from=build --chown=pythonic:pythonic /venv/fn /venv/fn
+RUN \
+  ln -sf /usr/local/bin/python3 /venv/fn/bin/python3
 EXPOSE 9443
-ENTRYPOINT ["/venv/fn/bin/pythonic"]
+ENTRYPOINT ["/venv/fn/bin/python", "-m", "crossplane.pythonic.main"]

crossplane/pythonic/function.py

@@ -5,6 +5,7 @@
 import builtins
 import importlib
 import inspect
+import sys
 
 import grpc
 import crossplane.function.logging
@@ -32,6 +33,14 @@ def __init__(self, debug=False):
         self.logger = crossplane.function.logging.get_logger()
         self.clazzes = {}
 
+    def invalidate_module(self, module):
+        if module in sys.modules:
+            del sys.modules[module]
+        for composite in [composite for composite in self.clazzes.keys()]:
+            if '\n' not in composite:
+                if composite.rsplit('.', 1)[0] == module:
+                    del self.clazzes[composite]
+
     async def RunFunction(
         self, request: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
     ) -> fnv1.RunFunctionResponse:
@@ -70,52 +79,52 @@ async def RunFunction(
                 try:
                     exec(composite, module.__dict__)
                 except Exception as e:
-                    crossplane.function.response.fatal(response, f"Exec exception: {e}")
                     logger.exception('Exec exception')
+                    crossplane.function.response.fatal(response, f"Exec exception: {e}")
                     return response
                 composite = ['<script>', 'Composite']
             else:
                 composite = composite.rsplit('.', 1)
                 if len(composite) == 1:
-                    crossplane.function.response.fatal(response, f"Composite class name does not include module: {composite[0]}")
                     logger.error(f"Composite class name does not include module: {composite[0]}")
+                    crossplane.function.response.fatal(response, f"Composite class name does not include module: {composite[0]}")
                     return response
                 try:
                     module = importlib.import_module(composite[0])
                 except Exception as e:
+                    logger.error(str(e))
                     crossplane.function.response.fatal(response, f"Import module exception: {e}")
-                    logger.exception('Import module exception')
                     return response
             clazz = getattr(module, composite[1], None)
             if not clazz:
-                crossplane.function.response.fatal(response, f"{composite[0]} did not define: {composite[1]}")
                 logger.error(f"{composite[0]} did not define: {composite[1]}")
+                crossplane.function.response.fatal(response, f"{composite[0]} did not define: {composite[1]}")
                 return response
             composite = '.'.join(composite)
             if not inspect.isclass(clazz):
-                crossplane.function.response.fatal(response, f"{composite} is not a class")
                 logger.error(f"{composite} is not a class")
+                crossplane.function.response.fatal(response, f"{composite} is not a class")
                 return response
             if not issubclass(clazz, BaseComposite):
-                crossplane.function.response.fatal(response, f"{composite} is not a subclass of BaseComposite")
                 logger.error(f"{composite} is not a subclass of BaseComposite")
+                crossplane.function.response.fatal(response, f"{composite} is not a subclass of BaseComposite")
                 return response
             self.clazzes[composite] = clazz
 
         try:
             composite = clazz(request, response, logger)
         except Exception as e:
-            crossplane.function.response.fatal(response, f"Instatiate exception: {e}")
             logger.exception('Instatiate exception')
+            crossplane.function.response.fatal(response, f"Instatiate exception: {e}")
             return response
 
         try:
             result = composite.compose()
             if asyncio.iscoroutine(result):
                 await result
         except Exception as e:
-            crossplane.function.response.fatal(response, f"Compose exception: {e}")
             logger.exception('Compose exception')
+            crossplane.function.response.fatal(response, f"Compose exception: {e}")
             return response
 
         unknownResources = []

crossplane/pythonic/main.py

@@ -1,19 +1,23 @@
 """The composition function's main CLI."""
 
-import warnings
-warnings.filterwarnings('ignore', module='^google[.]protobuf[.]runtime_version$', lineno=98)
-
 import argparse
+import asyncio
 import os
+import pathlib
 import shlex
+import signal
 import sys
+import traceback
+
+import crossplane.function.logging
+import crossplane.function.proto.v1.run_function_pb2_grpc as grpcv1
+import grpc
 import pip._internal.cli.main
-from crossplane.function import logging, runtime
 
 from . import function
 
 
-def main():
+async def main():
     parser = argparse.ArgumentParser('Forta Crossplane Function')
     parser.add_argument(
         '--debug', '-d',
@@ -27,20 +31,33 @@ def main():
     )
     parser.add_argument(
         '--tls-certs-dir',
+        default=os.getenv('TLS_SERVER_CERTS_DIR'),
         help='Serve using mTLS certificates.',
     )
     parser.add_argument(
         '--insecure',
         action='store_true',
         help='Run without mTLS credentials. If you supply this flag --tls-certs-dir will be ignored.',
     )
+    parser.add_argument(
+        '--packages',
+        action='store_true',
+        help='Discover python packages from function-pythonic ConfigMaps and Secrets.'
+    )
+    parser.add_argument(
+        '--packages-namespace',
+        action='append',
+        default=[],
+        help='Namespaces to discover function-pythonic ConfigMaps and Secrets in, default is cluster wide.',
+    )
     parser.add_argument(
         '--pip-install',
         help='Pip install command to install additional Python packages.'
     )
     parser.add_argument(
         '--python-path',
         action='append',
+        default=[],
         help='Filing system directories to add to the python path',
     )
     parser.add_argument(
@@ -49,33 +66,80 @@ def main():
         help='Allow oversized protobuf messages'
     )
     args = parser.parse_args()
-    if not args.tls_certs_dir:
-        args.tls_certs_dir = os.getenv('TLS_SERVER_CERTS_DIR')
+
+    if args.debug:
+        crossplane.function.logging.configure(crossplane.function.logging.Level.DEBUG)
+    else:
+        crossplane.function.logging.configure(crossplane.function.logging.Level.INFO)
 
     if args.pip_install:
         pip._internal.cli.main.main(['install', *shlex.split(args.pip_install)])
 
-    if args.python_path:
-        for path in reversed(args.python_path):
-            sys.path.insert(0, path)
+    # enables read only volumes or mismatched uid volumes
+    sys.dont_write_bytecode = True
+    for path in reversed(args.python_path):
+        sys.path.insert(0, path)
 
     if args.allow_oversize_protos:
         from google.protobuf.internal import api_implementation
         if api_implementation._c_module:
             api_implementation._c_module.SetAllowOversizeProtos(True)
 
-    logging.configure(logging.Level.DEBUG if args.debug else logging.Level.INFO)
-    runtime.serve(
-        function.FunctionRunner(args.debug),
-        args.address,
-        creds=runtime.load_credentials(args.tls_certs_dir),
-        insecure=args.insecure,
-    )
+    grpc.aio.init_grpc_aio()
+    grpc_runner = function.FunctionRunner(args.debug)
+    grpc_server = grpc.aio.server()
+    grpcv1.add_FunctionRunnerServiceServicer_to_server(grpc_runner, grpc_server)
+    if args.tls_certs_dir:
+        certs = pathlib.Path(args.tls_certs_dir)
+        grpc_server.add_secure_port(
+            args.address,
+            grpc.ssl_server_credentials(
+                private_key_certificate_chain_pairs=[(
+                    (certs / 'tls.key').read_bytes(),
+                    (certs / 'tls.crt').read_bytes(),
+                )],
+                root_certificates=(certs / 'ca.crt').read_bytes(),
+                require_client_auth=True,
+            ),
+        )
+    else:
+        if not args.insecure:
+            raise ValueError('Either --tls-certs-dir or --insecure must be specified')
+        grpc_server.add_insecure_port(args.address)
+    await grpc_server.start()
+
+    if args.packages:
+        import kopf._core.actions.loggers
+        import kopf._core.reactor.running
+        from . import packages
+        sys.path.insert(0, str(packages.PACKAGES_DIR))
+        packages.register_grpc_runner(grpc_runner)
+        kopf._core.actions.loggers.configure()
+        @kopf.on.startup()
+        async def startup(settings, **_):
+            settings.scanning.disabled = True
+        @kopf.on.cleanup()
+        async def cleanup(logger=None, **_):
+            await grpc_server.stop(5)
+        async with asyncio.TaskGroup() as tasks:
+            tasks.create_task(grpc_server.wait_for_termination())
+            tasks.create_task(kopf._core.reactor.running.operator(
+                standalone=True,
+                clusterwide=not args.packages_namespace,
+                namespaces=args.packages_namespace,
+            ))
+    else:
+        def stop():
+            asyncio.ensure_future(grpc_server.stop(5))
+        loop = asyncio.get_event_loop()
+        loop.add_signal_handler(signal.SIGINT, stop)
+        loop.add_signal_handler(signal.SIGTERM, stop)
+        await grpc_server.wait_for_termination()
 
 
 if __name__ == "__main__":
     try:
-        main()
-    except Exception as e:
-        print(f"Exception running main: {e}", file=sys.stderr)
+        asyncio.run(main())
+    except:
+        print(traceback.format_exc())
         sys.exit(1)