Merge pull request #139 from gefyrahq/feat/#138

feat(client): read endpoint connection from kubeconfig

client/gefyra/__main__.py

@@ -4,7 +4,11 @@
 
 from gefyra.api import get_containers_and_print, get_bridges_and_print
 from gefyra.configuration import ClientConfiguration
-from gefyra.local.utils import PortMappingParser, IpPortMappingParser
+from gefyra.local.utils import (
+    PortMappingParser,
+    IpPortMappingParser,
+    get_connection_from_kubeconfig,
+)
 from gefyra.local.minikube import detect_minikube_config
 from gefyra.local.telemetry import CliTelemetry
 
@@ -204,8 +208,15 @@ def up_command(args):
     if args.minikube:
         configuration = detect_minikube_config(args)
     else:
+        if not args.endpoint:
+            # #138: Read in the --endpoint parameter from kubeconf
+            endpoint = get_connection_from_kubeconfig()
+            logger.info(f"Setting --endpoint from kubeconfig {endpoint}")
+        else:
+            endpoint = args.endpoint
+
         configuration = ClientConfiguration(
-            cargo_endpoint=args.endpoint,
+            cargo_endpoint=endpoint,
             registry_url=args.registry,
             operator_image_url=args.operator,
             stowaway_image_url=args.stowaway,

client/gefyra/local/utils.py

@@ -165,6 +165,28 @@ def handle_docker_run_container(
     )
 
 
+def get_connection_from_kubeconfig() -> Optional[str]:
+    from kubernetes.config.kube_config import KUBE_CONFIG_DEFAULT_LOCATION
+    from pathlib import Path
+    import yaml
+
+    try:
+        with open(Path(KUBE_CONFIG_DEFAULT_LOCATION).expanduser(), "r") as kubeconfig:
+            kubecfg = yaml.safe_load(kubeconfig)
+        active_ctx = next(
+            filter(
+                lambda x: x["name"] == kubecfg["current-context"], kubecfg["contexts"]
+            )
+        )
+        if gefyra_connection := active_ctx.get("gefyra"):
+            return gefyra_connection
+        else:
+            return None
+    except Exception as e:  # noqa
+        logger.error(f"Could not load Gefyra --endpoint from kubeconfig due to: {e}")
+        return None
+
+
 class PortMappingParser(argparse.Action):
     """Adapted from https://stackoverflow.com/questions/29986185/python-argparse-dict-arg"""
 

client/tests/test_cli_utils.py

@@ -0,0 +1,42 @@
+import os
+from unittest.mock import patch
+
+import yaml
+
+from gefyra.local.utils import get_connection_from_kubeconfig
+
+
+@patch("kubernetes.config.kube_config.KUBE_CONFIG_DEFAULT_LOCATION", "/tmp/kube.yaml")
+def test_get_connection_from_kubeconfig_no_connection():
+    endpoint = get_connection_from_kubeconfig()
+    assert endpoint is None
+
+
+@patch("kubernetes.config.kube_config.KUBE_CONFIG_DEFAULT_LOCATION", "/tmp/kube.yaml")
+def test_get_connection_from_kubeconfig_connection():
+    data = {
+        "current-context": "fake",
+        "contexts": [
+            {
+                "name": "fake",
+                "gefyra": "127.0.0.1:8090",
+            },
+        ],
+    }
+    f = open("/tmp/kube.yaml", "w")
+    yaml.dump(data, f)
+    f.close()
+    try:
+        endpoint = get_connection_from_kubeconfig()
+        assert endpoint == "127.0.0.1:8090"
+    except AssertionError:
+        os.remove(f.name)
+        raise
+    else:
+        os.remove(f.name)
+
+
+@patch("kubernetes.config.kube_config.KUBE_CONFIG_DEFAULT_LOCATION", "/tmp/kube1.yaml")
+def test_get_connection_from_kubeconfig_no_file():
+    endpoint = get_connection_from_kubeconfig()
+    assert endpoint is None