Implement ConfigMap and Secret based python packages.

Patrick J. McNerthney · Patrick J. McNerthney · commit 2f469256d596 · 2025-08-15T13:00:03.000-10:00
diff --git a/Dockerfile b/Dockerfile
@@ -40,4 +40,4 @@ WORKDIR /
 USER nonroot:nonroot
 COPY --from=build --chown=nonroot:nonroot /venv/fn /venv/fn
 EXPOSE 9443
-ENTRYPOINT ["/venv/fn/bin/pythonic"]
+ENTRYPOINT ["/venv/fn/bin/python", "-m", "crossplane.pythonic.main"]
diff --git a/crossplane/pythonic/function.py b/crossplane/pythonic/function.py
@@ -32,6 +32,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 +78,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 = []
diff --git a/crossplane/pythonic/main.py b/crossplane/pythonic/main.py
@@ -1,19 +1,24 @@
 """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 crossplane.function.logging
+import crossplane.function.proto.v1.run_function_pb2_grpc as grpcv1
+import grpc
+import kopf._core.actions.loggers
+import kopf._core.reactor.running
 import pip._internal.cli.main
-from crossplane.function import logging, runtime
 
-from . import function
+from . import function, packages
 
 
-def main():
+async def main():
     parser = argparse.ArgumentParser('Forta Crossplane Function')
     parser.add_argument(
         '--debug', '-d',
@@ -27,13 +32,25 @@ 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.'
@@ -49,8 +66,6 @@ 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.pip_install:
         pip._internal.cli.main.main(['install', *shlex.split(args.pip_install)])
@@ -64,18 +79,62 @@ def main():
         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,
-    )
+    if args.debug:
+        crossplane.function.logging.configure(crossplane.function.logging.Level.DEBUG)
+    else:
+        crossplane.function.logging.configure(crossplane.function.logging.Level.INFO)
+
+    grpc_runner = function.FunctionRunner(args.debug)
+    packages.register_grpc_runner(grpc_runner)
+    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)
+
+    if args.packages:
+        sys.path.insert(0, str(packages.PACKAGES_DIR))
+        sys.dont_write_bytecode = True
+        kopf._core.actions.loggers.configure()
+        @kopf.on.cleanup()
+        async def cleanup(logger=None, **_):
+            await grpc_server.stop(5)
+    else:
+        def cleanup():
+            asyncio.ensure_future(grpc_server.stop(5))
+        loop = asyncio.get_event_loop()
+        loop.add_signal_handler(signal.SIGINT, cleanup)
+        loop.add_signal_handler(signal.SIGTERM, cleanup)
+
+    async with asyncio.TaskGroup() as tasks:
+        grpc.aio.init_grpc_aio()
+        await grpc_server.start()
+        tasks.create_task(grpc_server.wait_for_termination())
+        if args.packages:
+            tasks.create_task(kopf._core.reactor.running.operator(
+                standalone=True,
+                clusterwide=not args.packages_namespace,
+                namespaces=args.packages_namespace,
+            ))
 
 
 if __name__ == "__main__":
     try:
-        main()
+        asyncio.run(main())
     except Exception as e:
         print(f"Exception running main: {e}", file=sys.stderr)
         sys.exit(1)
diff --git a/crossplane/pythonic/packages.py b/crossplane/pythonic/packages.py
@@ -0,0 +1,138 @@
+
+import base64
+import importlib
+import pathlib
+import sys
+
+import kopf
+
+
+PACKAGES_DIR = pathlib.Path(sys.prefix) / 'lib' / 'function-pythonic'
+GRPC_RUNNER = None
+
+
+@kopf.on.create('', 'v1', 'configmaps', labels={'function-pythonic.package': kopf.PRESENT})
+@kopf.on.resume('', 'v1', 'configmaps', labels={'function-pythonic.package': kopf.PRESENT})
+@kopf.on.create('', 'v1', 'secrets', labels={'function-pythonic.package': kopf.PRESENT})
+@kopf.on.resume('', 'v1', 'secrets', labels={'function-pythonic.package': kopf.PRESENT})
+async def create(body, logger, **_):
+    package_dir, package = get_package_dir(body)
+    if package_dir:
+        package_dir.mkdir(parents=True, exist_ok=True)
+        secret = body['kind'] == 'Secret'
+        invalidate = False
+        for name, text in body.get('data', {}).items():
+            package_file = package_dir / name
+            if secret:
+                package_file.write_bytes(base64.b64decode(text.encode('utf-8')))
+            else:
+                package_file.write_text(text)
+            if package_file.suffixes == ['.py']:
+                logger.info(f"Created module: {'.'.join(package + [package_file.stem])}")
+                invalidate = True
+            else:
+                logger.info(f"Created file: {'/'.join(package + [name])}")
+        if invalidate:
+            importlib.invalidate_caches()
+
+
+@kopf.on.update('', 'v1', 'configmaps', labels={'function-pythonic.package': kopf.PRESENT})
+@kopf.on.update('', 'v1', 'secrets', labels={'function-pythonic.package': kopf.PRESENT})
+async def update(body, old, logger, **_):
+    old_package_dir, old_package = get_package_dir(old)
+    if old_package_dir:
+        old_data = old.get('data', {})
+    else:
+        old_data = {}
+    old_names = set(old_data.keys())
+    package_dir, package = get_package_dir(body, logger)
+    if package_dir:
+        package_dir.mkdir(parents=True, exist_ok=True)
+        secret = body['kind'] == 'Secret'
+        invalidate = False
+        for name, text in body.get('data', {}).items():
+            if package_dir == old_package_dir and text == old_data.get(name, None):
+                old_names.discard(name)
+            else:
+                package_file = package_dir / name
+                if secret:
+                    package_file.write_bytes(base64.b64decode(text.encode('utf-8')))
+                else:
+                    package_file.write_text(text)
+                if package_file.suffixes == ['.py']:
+                    module = '.'.join(package + [package_file.stem])
+                    if name in old_names:
+                        GRPC_RUNNER.invalidate_module(module)
+                        old_names.discard(name)
+                    logger.info(f"Updated module: {module}")
+                    invalidate = True
+                else:
+                    logger.info(f"Updated file: {'/'.join(old_package + [name])}")
+        if invalidate:
+            importlib.invalidate_caches()
+    if old_package_dir:
+        for name in old_names:
+            package_file = old_package_dir / name
+            package_file.unlink(missing_ok=True)
+            if package_file.suffixes == ['.py']:
+                module = '.'.join(old_package + [package_file.stem])
+                GRPC_RUNNER.invalidate_module(module)
+                logger.info(f"Removed module: {module}")
+            else:
+                logger.info(f"Removed file: {'/'.join(old_package + [name])}")
+            logger.info('update deleted: %s', package_file)
+        while old_package and old_package_dir.is_dir() and not list(old_package_dir.iterdir()):
+            old_package_dir.rmdir()
+            module = '.'.join(old_package)
+            GRPC_RUNNER.invalidate_module(module)
+            logger.info(f"Removed package: {module}")
+            old_package_dir = old_package_dir.parent
+            old_package.pop()
+
+
+@kopf.on.delete('', 'v1', 'configmaps', labels={'function-pythonic.package': kopf.PRESENT})
+@kopf.on.delete('', 'v1', 'secrets', labels={'function-pythonic.package': kopf.PRESENT})
+async def delete(old, logger, **_):
+    package_dir, package = get_package_dir(old)
+    if package_dir:
+        for name in old.get('data', {}).keys():
+            package_file = package_dir / name
+            package_file.unlink(missing_ok=True)
+            if package_file.suffixes == ['.py']:
+                module = '.'.join(package + [package_file.stem])
+                GRPC_RUNNER.invalidate_module(module)
+                logger.info(f"Deleted module: {module}")
+            else:
+                logger.info(f"Deleted file: {'/'.join(package + [name])}")
+        while package and package_dir.is_dir() and not list(package_dir.iterdir()):
+            package_dir.rmdir()
+            module = '.'.join(package)
+            GRPC_RUNNER.invalidate_module(module)
+            logger.info(f"Deleted package: {module}")
+            package_dir = package_dir.parent
+            package.pop()
+
+
+def register_grpc_runner(runner):
+    global GRPC_RUNNER
+    GRPC_RUNNER = runner
+
+
+def get_package_dir(body, logger=None):
+    package = body.get('metadata', {}).get('labels', {}).get('function-pythonic.package', None)
+    if package is None:
+        if logger:
+            logger.error('function-pythonic.package label is missing')
+        return None, None
+    package_dir = PACKAGES_DIR
+    if package == '':
+        package = []
+    else:
+        package = package.split('.')
+        for segment in package:
+            if not segment.isidentifier():
+                if logger:
+                    logger.error('Package has invalid package name: %s', package)
+                return None, None
+            package_dir = package_dir / segment
+    return package_dir, package
diff --git a/examples/eks-cluster/functions.yaml b/examples/eks-cluster/functions.yaml
@@ -2,7 +2,7 @@ apiVersion: pkg.crossplane.io/v1beta1
 kind: Function
 metadata:
   name: function-pythonic
-  #annotations: 
-  #  render.crossplane.io/runtime: Development
+  annotations:
+    render.crossplane.io/runtime: Development
 spec:
   package: ghcr.io/fortra/function-pythonic:v0.0.5
diff --git a/pyproject.toml b/pyproject.toml
@@ -19,6 +19,7 @@ classifiers = [
 
 dependencies = [
   "crossplane-function-sdk-python==0.9.0",
+  "kopf==1.38.0",
   "pyyaml==6.0.2",
 ]
 
@@ -29,9 +30,6 @@ Documentation = "https://github.com/fortra/function-pythonic#readme"
 Issues = "https://github.com/fortra/function-pythonic/issues"
 Source = "https://github.com/fortra/function-pythonic"
 
-[project.scripts]
-pythonic = "crossplane.pythonic.main:main"
-
 [tool.hatch.build.targets.wheel]
 packages = ["crossplane"]
 
@@ -44,8 +42,9 @@ type = "virtual"
 path = ".venv-default"
 dependencies = ["ipython==9.1.0"]
 [tool.hatch.envs.default.scripts]
-development = "python -m crossplane.pythonic.main --insecure --debug"
 production = "python -m crossplane.pythonic.main --insecure"
+development = "python -m crossplane.pythonic.main --insecure --debug"
+packages = "python -m crossplane.pythonic.main --insecure --debug --packages"
 
 [tool.hatch.envs.lint]
 type = "virtual"
diff --git a/scripts/setup-local.sh b/scripts/setup-local.sh
diff --git a/tests/test_fn.py b/tests/test_fn.py
