Merge pull request #16 from karellen/dynamic_k8s_client

Add dynamic selection of K8S python client based on connection version

Dockerfile

@@ -9,4 +9,5 @@ RUN pip install --no-input --no-cache-dir /tmp/*.whl && \
     rm -rf /tmp/*
 
 WORKDIR /root
+RUN kubernator --pre-cache-k8s-client $(seq 19 28)
 ENTRYPOINT ["/usr/local/bin/kubernator"]

build.py

@@ -27,7 +27,7 @@
 use_plugin("filter_resources")
 
 name = "kubernator"
-version = "1.0.5"
+version = "1.0.6.dev"
 
 summary = "Kubernator is the a pluggable framework for K8S provisioning"
 authors = [Author("Express Systems USA, Inc.", "")]

src/integrationtest/python/smoke_clear_cache_tests.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+#
+#   Copyright 2020 Express Systems USA, Inc
+#   Copyright 2023 Karellen, Inc.
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+#
+
+from test_support import IntegrationTestSupport
+
+
+class VersionSmokeTest(IntegrationTestSupport):
+    def test_version(self):
+        self.smoke_test_module("kubernator", "--clear-cache")

src/integrationtest/python/smoke_version_tests.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+#
+#   Copyright 2020 Express Systems USA, Inc
+#   Copyright 2023 Karellen, Inc.
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+#
+
+from test_support import IntegrationTestSupport
+
+
+class VersionSmokeTest(IntegrationTestSupport):
+    def test_version(self):
+        self.smoke_test_module("kubernator", "--version")

src/integrationtest/python/test_support.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+#
+#   Copyright 2020 Express Systems USA, Inc
+#   Copyright 2023 Karellen, Inc.
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+#
+
+import sys
+from os import chdir, getcwd
+from runpy import run_module
+from unittest import TestCase
+
+
+class IntegrationTestSupport(TestCase):
+
+    def smoke_test_module(self, module, *args):
+        old_argv = list(sys.argv)
+        del sys.argv[:]
+        sys.argv.append("bogus")
+        sys.argv.extend(args)
+
+        old_modules = dict(sys.modules)
+        old_meta_path = list(sys.meta_path)
+        old_sys_path = list(sys.path)
+        old_cwd = getcwd()
+        # chdir(self.tmp_directory)
+        try:
+            return run_module(module, run_name="__main__")
+        except SystemExit as e:
+            self.assertEqual(e.code, 0, "Test did not exit successfully: %r" % e.code)
+        finally:
+            del sys.argv[:]
+            sys.argv.extend(old_argv)
+
+            sys.modules.clear()
+            sys.modules.update(old_modules)
+
+            del sys.meta_path[:]
+            sys.meta_path.extend(old_meta_path)
+
+            del sys.path[:]
+            sys.path.extend(old_sys_path)
+
+            chdir(old_cwd)

src/main/python/kubernator/__init__.py

@@ -21,7 +21,7 @@
 
 def _main():
     from kubernator import app
-    return app.main()
+    return app.main() or 0
 
 
 def main():

src/main/python/kubernator/api.py

@@ -152,8 +152,12 @@ def _download_remote_file(url, file_name, cache: dict):
             return dict(r.headers)
 
 
+def get_app_cache_dir():
+    return Path(user_config_dir("kubernator"))
+
+
 def get_cache_dir(category: str, sub_category: str = None):
-    config_dir = Path(user_config_dir("kubernator")) / category
+    config_dir = get_app_cache_dir() / category
     if sub_category:
         config_dir = config_dir / sub_category
     if not config_dir.exists():
@@ -224,6 +228,19 @@ def set_defaults(validator, properties, instance, schema):
     return validators.extend(validator_class, {"properties": set_defaults})
 
 
+def install_python_k8s_client(run, package_major, logger_stdout, logger_stderr):
+    cache_dir = get_cache_dir("python")
+    package_major_dir = cache_dir / str(package_major)
+
+    if not package_major_dir.exists():
+        package_major_dir.mkdir(parents=True, exist_ok=True)
+
+        run(["pip", "install", "--no-deps", "--no-cache-dir", "--no-input", "--target", str(package_major_dir),
+             f"kubernetes~={package_major}.0"], logger_stdout, logger_stderr).wait()
+
+    return package_major_dir
+
+
 class _PropertyList(MutableSequence):
 
     def __init__(self, seq, read_parent, name):

src/main/python/kubernator/app.py

@@ -26,11 +26,13 @@
 from collections import deque
 from collections.abc import MutableMapping, Callable
 from pathlib import Path
+from shutil import rmtree
 from typing import Optional, Union
 
 import kubernator
 from kubernator.api import (KubernatorPlugin, Globs, scan_dir, PropertyDict, config_as_dict, config_parent,
-                            download_remote_file, load_remote_file, Repository, StripNL, jp)
+                            download_remote_file, load_remote_file, Repository, StripNL, jp, get_app_cache_dir,
+                            install_python_k8s_client)
 from kubernator.proc import run, run_capturing_out
 
 TRACE = 5
@@ -55,10 +57,16 @@ def trace(self, msg, *args, **kwargs):
 
 
 def define_arg_parse():
-    parser = argparse.ArgumentParser(description="Kubernetes Provisioning Tool",
+    parser = argparse.ArgumentParser(prog="kubernator",
+                                     description="Kubernetes Provisioning Tool",
                                      formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    parser.add_argument("--version", action="version", version=kubernator.__version__,
-                        help="print version and exit")
+    g = parser.add_mutually_exclusive_group()
+    g.add_argument("--version", action="version", version=kubernator.__version__,
+                   help="print version and exit")
+    g.add_argument("--clear-cache", action="store_true",
+                   help="clear cache and exit")
+    g.add_argument("--pre-cache-k8s-client", action="extend", nargs="+", type=int,
+                   help="download specified K8S client library minor(!) version(s) and exit")
     parser.add_argument("--log-format", choices=["human", "json"], default="human",
                         help="whether to log for human or machine consumption")
     parser.add_argument("--log-file", type=argparse.FileType("w"), default=None,
@@ -440,11 +448,36 @@ def __repr__(self):
         return "Kubernator"
 
 
+def clear_cache():
+    cache_dir = get_app_cache_dir()
+    logger.info("Clearing application cache at %s", cache_dir)
+    if cache_dir.exists():
+        rmtree(cache_dir)
+
+
+def pre_cache_k8s_clients(*versions):
+    proc_logger = logger.getChild("proc")
+    stdout_logger = StripNL(proc_logger.info)
+    stderr_logger = StripNL(proc_logger.warning)
+
+    for v in versions:
+        logger.info("Caching K8S client library ~=v%s.0...", v)
+        install_python_k8s_client(run, v, stdout_logger, stderr_logger)
+
+
 def main():
     args = define_arg_parse().parse_args()
     init_logging(args.verbose, args.log_format, args.log_file)
 
     try:
+        if args.clear_cache:
+            clear_cache()
+            return
+
+        if args.pre_cache_k8s_client:
+            pre_cache_k8s_clients(*args.pre_cache_k8s_client)
+            return
+
         with App(args) as app:
             app.run()
     except SystemExit as e:

src/main/python/kubernator/plugins/istio.py

@@ -145,7 +145,8 @@ def handle_start(self):
         # This plugin only deals with Istio Operator, so only load that stuff
         self.resource_definitions_schema = load_remote_file(logger,
                                                             f"https://raw.githubusercontent.com/kubernetes/kubernetes/"
-                                                            f"{self.context.k8s.server_version}/api/openapi-spec/swagger.json",
+                                                            f"{self.context.k8s.server_git_version}"
+                                                            f"/api/openapi-spec/swagger.json",
                                                             FileType.JSON)
         self._populate_resource_definitions()
         self.add_remote_crds(f"{url_prefix}/crd-operator.yaml", FileType.YAML)

src/main/python/kubernator/plugins/k8s.py

@@ -24,19 +24,30 @@
 import types
 from collections.abc import Mapping
 from functools import partial
+from importlib.metadata import version as pkg_version
 from pathlib import Path
 from typing import Iterable, Callable, Sequence
 
 import jsonpatch
 import yaml
 
-from kubernator.api import (KubernatorPlugin, Globs, scan_dir, load_file, FileType, load_remote_file)
+from kubernator.api import (KubernatorPlugin,
+                            Globs,
+                            scan_dir,
+                            load_file,
+                            FileType,
+                            load_remote_file,
+                            StripNL,
+                            install_python_k8s_client)
 from kubernator.plugins.k8s_api import (K8SResourcePluginMixin,
                                         K8SResource,
                                         K8SResourcePatchType,
                                         K8SPropagationPolicy)
 
 logger = logging.getLogger("kubernator.k8s")
+proc_logger = logger.getChild("proc")
+stdout_logger = StripNL(proc_logger.info)
+stderr_logger = StripNL(proc_logger.warning)
 
 
 def final_resource_validator(resources: Sequence[K8SResource],
@@ -62,6 +73,8 @@ def __init__(self):
         super().__init__()
         self.context = None
 
+        self.embedded_pkg_version = self._get_kubernetes_client_version()
+
         self._transformers = []
         self._validators = []
 
@@ -109,30 +122,63 @@ def handle_start(self):
     def _kubeconfig_changed(self):
         self.setup_client()
 
+    def _get_kubernetes_client_version(self):
+        return pkg_version("kubernetes").split(".")
+
     def setup_client(self):
+        k8s = self.context.k8s
+        if "server_version" not in k8s:
+            self._setup_client()
+
+        server_minor = k8s.server_version[1]
+        pkg_major = self.embedded_pkg_version[0]
+        if server_minor != pkg_major:
+            logger.info("Bundled Kubernetes client version %s doesn't match server version %s",
+                        ".".join(self.embedded_pkg_version), ".".join(k8s.server_version))
+            pkg_dir = install_python_k8s_client(self.context.app.run, server_minor, stdout_logger, stderr_logger)
+
+            modules_to_delete = []
+            for k, v in sys.modules.items():
+                if k == "kubernetes" or k.startswith("kubernetes."):
+                    modules_to_delete.append(k)
+            for k in modules_to_delete:
+                del sys.modules[k]
+
+            logger.info("Adding sys.path reference to %s", pkg_dir)
+            sys.path.insert(0, str(pkg_dir))
+            self.embedded_pkg_version = self._get_kubernetes_client_version()
+            logger.info("Switching to Kubernetes client version %s", ".".join(self.embedded_pkg_version))
+            self._setup_client()
+        else:
+            logger.info("Bundled Kubernetes client version %s matches server version %s",
+                        ".".join(self.embedded_pkg_version), ".".join(k8s.server_version))
+
+        k8s_def = load_remote_file(logger, f"https://raw.githubusercontent.com/kubernetes/kubernetes/"
+                                           f"{k8s.server_git_version}/api/openapi-spec/swagger.json",
+                                   FileType.JSON)
+        self.resource_definitions_schema = k8s_def
+
+        self._populate_resource_definitions()
+
+    def _setup_client(self):
         from kubernetes import client
 
         context = self.context
+        k8s = context.k8s
 
-        context.k8s.client = self._setup_k8s_client()
-        version = client.VersionApi(context.k8s.client).get_code()
+        k8s.client = self._setup_k8s_client()
+        version = client.VersionApi(k8s.client).get_code()
         if "-eks-" in version.git_version:
             git_version = version.git_version.split("-")[0]
         else:
             git_version = version.git_version
 
-        context.k8s.server_version = git_version
-
-        logger.info("Found Kubernetes %s on %s", version.git_version, context.k8s.client.configuration.host)
-
-        logger.debug("Reading Kubernetes OpenAPI spec for version %s", git_version)
+        k8s.server_version = git_version[1:].split(".")
+        k8s.server_git_version = git_version
 
-        k8s_def = load_remote_file(logger, f"https://raw.githubusercontent.com/kubernetes/kubernetes/"
-                                           f"{git_version}/api/openapi-spec/swagger.json",
-                                   FileType.JSON)
-        self.resource_definitions_schema = k8s_def
+        logger.info("Found Kubernetes %s on %s", k8s.server_git_version, k8s.client.configuration.host)
 
-        self._populate_resource_definitions()
+        logger.debug("Reading Kubernetes OpenAPI spec for %s", k8s.server_git_version)
 
     def handle_before_dir(self, cwd: Path):
         context = self.context