Implement ConfigMap and Secret based python packages.

Author: Patrick J. McNerthney

.gitignore

@@ -209,3 +209,6 @@ cython_debug/
 marimo/_static/
 marimo/_lsp/
 __marimo__/
+
+# function-pythonic
+pythonic-packages/

Dockerfile

@@ -1,43 +1,20 @@
-# syntax=docker/dockerfile:1
+FROM python:3.13-slim-trixie AS image
 
-# It's important that this is Debian 12 to match the distroless image.
-FROM debian:12-slim AS build
-
-#RUN --mount=type=cache,target=/var/lib/apt/lists \
-#    --mount=type=cache,target=/var/cache/apt \
-RUN \
-    rm -f /etc/apt/apt.conf.d/docker-clean \
-    && apt-get update \
-    && apt-get install --no-install-recommends --yes python3-venv git
-
-# Don't write .pyc bytecode files. These speed up imports when the program is
-# loaded. There's no point doing that in a container where they'll never be
-# persisted across restarts.
-ENV PYTHONDONTWRITEBYTECODE=true
-
-# Use Hatch to build a wheel. The build stage must do this in a venv because
-# Debian doesn't have a hatch package, and it won't let you install one globally
-# using pip.
-WORKDIR /build
-#RUN --mount=target=. \
-#    --mount=type=cache,target=/root/.cache/pip \
-COPY . /build
-RUN \
-    python3 -m venv /venv/build \
-    && /venv/build/bin/pip install hatch \
-    && /venv/build/bin/hatch build -t wheel /whl
-
-# Create a fresh venv and install only the pythonic wheel into it.
-#RUN --mount=type=cache,target=/root/.cache/pip \
+WORKDIR /root/pythonic
+COPY pyproject.toml /root/pythonic
+COPY crossplane /root/pythonic/crossplane
+WORKDIR /
 RUN \
-    python3 -m venv /venv/fn \
-    && /venv/fn/bin/pip install /whl/*.whl
+  set -eux && \
+  cd /root/pythonic && \
+  pip install --root-user-action ignore --no-build-isolation setuptools==80.9.0 && \
+  pip install --root-user-action ignore --no-build-isolation . && \
+  cd .. && \
+  rm -rf .cache pythonic && \
+  groupadd --gid 2000 pythonic && \
+  useradd --uid 2000 --gid pythonic --home-dir /opt/pythonic --create-home --shell /usr/sbin/nologin pythonic
 
-# 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
+USER pythonic:pythonic
+WORKDIR /opt/pythonic
 EXPOSE 9443
-ENTRYPOINT ["/venv/fn/bin/pythonic"]
+ENTRYPOINT ["python", "-m", "crossplane.pythonic.main"]

crossplane/pythonic/__version__.py

@@ -5,6 +5,7 @@
 import builtins
 import importlib
 import inspect
+import sys
 
 import grpc
 import crossplane.function.logging
@@ -32,6 +33,12 @@ def __init__(self, debug=False):
         self.logger = crossplane.function.logging.get_logger()
         self.clazzes = {}
 
+    def invalidate_module(self, module):
+        self.clazzes.clear()
+        if module in sys.modules:
+            del sys.modules[module]
+        importlib.invalidate_caches()
+
     async def RunFunction(
         self, request: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
     ) -> fnv1.RunFunctionResponse:
@@ -70,52 +77,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/function.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,38 @@ 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(
+        '--packages-dir',
+        default='./pythonic-packages',
+        help='Directory to store discovered function-pythonic ConfigMaps and Secrets to, defaults "<cwd>/pythonic-packages"'
+    )
     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 +71,81 @@ 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, str(pathlib.Path(path).resolve()))
 
     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
+        packages_dir = pathlib.Path(args.packages_dir).expanduser().resolve()
+        sys.path.insert(0, str(packages_dir))
+        packages.setup(packages_dir, 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)