feat: try loading k8s from outside cluster

README.md

@@ -29,12 +29,13 @@ To use QScaler-SDK, configure the following environment variables:
 
 ### Environment Variables
 
-| Variable           | Description                                                                                   |
-|--------------------|-----------------------------------------------------------------------------------------------|
-| `QWORKER_NAME`     | Name of the QWorker CRD to monitor.                                                          |
-| `PULLING_INTERVAL` | Interval (in seconds) for checking the QWorker CRD status when no tasks are running.          |
-| `POD_SPEC_HASH`    | Hash of the current pod specification. Used to detect changes in the CRD and trigger shutdown.|
-| `HOSTNAME`         | Name of the pod, automatically set by Kubernetes.                                            |
+| Variable           | Description                                                                                    |
+|--------------------|------------------------------------------------------------------------------------------------|
+| `QWORKER_NAME`     | Name of the QWorker CRD to monitor.                                                            |
+| `PULLING_INTERVAL` | Interval (in seconds) for checking the QWorker CRD status when no tasks are running.           |
+| `POD_SPEC_HASH`    | Hash of the current pod specification. Used to detect changes in the CRD and trigger shutdown. |
+| `HOSTNAME`         | Name of the pod, automatically set by Kubernetes.                                              |
+| `KUBECONFIG`       | KubeConfig file path, used if running from outside of k8s cluster.                              |
 
 ## Example
 

conftest.py

@@ -12,9 +12,8 @@ def mock_incluster_environment(request, mocker):
     if "incluster" in request.keywords:
         os.environ.setdefault("QWORKER_NAME", "fake-qworker")
         os.environ.setdefault("HOSTNAME", "localhost")
-        mocker.patch("qscaler.k8s.k8s_client.K8sClient._load_namespace_from_file", return_value="fake")
-        mocker.patch("qscaler.k8s.k8s_client.cluster_config.load_incluster_config", return_value=None)
-
+        mocker.patch("qscaler.k8s.k8s_client.K8sClient._load_namespace", return_value="fake")
+        mocker.patch("qscaler.k8s.k8s_client.cluster_config.load_config", return_value=None)
 
 
 @pytest.fixture(autouse=True)

qscaler/configuration/config.py

@@ -8,6 +8,6 @@ class Config:
     pulling_interval: int = field(default=os.getenv("PULLING_INTERVAL", 2))
     pod_name: str = field(default=os.getenv("HOSTNAME"))  # in K8s it is the name of the pod
     pod_spec_hash: str = field(default=os.getenv("POD_SPEC_HASH"))
-
+    kubeconfig: str = field(default=os.getenv("KUBECONFIG"))
 
 config = Config()

qscaler/k8s/k8s_client.py

@@ -1,10 +1,12 @@
 import logging
+import os
 from base64 import b64decode
 from typing import Any
 
 from kubernetes import client, config as cluster_config
 from kubernetes.client.rest import ApiException
 
+from qscaler.configuration.config import config
 from qscaler.utils.singleton import SingletonMeta
 
 logger = logging.getLogger(__name__)
@@ -14,18 +16,22 @@
 class K8sClient(metaclass=SingletonMeta):
 
     def __init__(self):
-        cluster_config.load_incluster_config()
+        cluster_config.load_config()
         self.api_group = "quickube.com"
         self.api_version = "v1alpha1"
-        self.namespace = self._load_namespace_from_file()
+        self.namespace = self._load_namespace()
 
     @staticmethod
-    def _load_namespace_from_file():
-        try:
-            with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r") as f:
+    def _load_namespace() -> str | None:
+        ns_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+        if os.path.exists(ns_path):
+            with open(ns_path) as f:
                 return f.read().strip()
-        except FileNotFoundError:
-            raise RuntimeError("Namespace file not found. Are you running in a Kubernetes cluster?")
+        try:
+            contexts, active_context = cluster_config.list_kube_config_contexts()
+            return active_context["context"]["namespace"]
+        except (KeyError, StopIteration):
+            return "default"
 
     def get_qworker(self, name: str) -> dict:
         api_instance = client.CustomObjectsApi()