Skip to content

fix: incluster config handling #2540

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: incluster config handling #2540

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 32 additions & 25 deletions ocp_resources/resource.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
TimeoutSampler,
TimeoutWatch,
)
from urllib3.exceptions import MaxRetryError

from fake_kubernetes_client.dynamic_client import FakeDynamicClient
from ocp_resources.event import Event
Expand Down Expand Up @@ -204,16 +203,20 @@ def get_client(
token: str | None = None,
fake: bool = False,
) -> DynamicClient | FakeDynamicClient:
"""
Get a kubernetes client.
"""Get a kubernetes client.

Valid combinations for arguments include:
- config file (kubeconfig path)
- config_dict
- username, password and host (uses basic auth)
- host and token (bearer token)
- client_configuration

This function is a replica of `ocp_utilities.infra.get_client` which cannot be imported as ocp_utilities imports
from ocp_resources.
If none of the above arguments are provided:
- if the `KUBECONFIG` env var is set: it will be used to configure the client
- if not set and `~/.kube/config` exists, it will be used to configure the client

Pass either config_file or config_dict.
If none of them are passed, client will be created from default OS kubeconfig
(environment variable or .kube folder).
If no configuration files are available, incluster config loading will be attempted

Args:
config_file (str): path to a kubeconfig file.
Expand Down Expand Up @@ -264,34 +267,38 @@ def get_client(
persist_config=persist_config,
temp_file_path=temp_file_path,
)
else:
# Ref: https://github.com/kubernetes-client/python/blob/v26.1.0/kubernetes/base/config/__init__.py
LOGGER.info("Trying to get client via new_client_from_config")

elif config_file or ("KUBECONFIG" in os.environ) or os.path.exists(os.path.expanduser("~/.kube/config")):
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

("KUBECONFIG" in os.environ) this is not a good check, this will be true for TEST_KUBECONFIG in os.environ.
Use os.environ.get("KUBECONFIG") and save it since you need to use it later.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

TEST_KUBECONFIG as an environment variable? In that case "KUBECONFIG" in os.environ will not be true, os.environ is a dict and will not have the "KUBECONFIG" key.

The idea here is to go through this path only if:

  • config_file is passed by the user explicitly
  • user explicitly set a KUBECONFIG environment variable (so the key will be in os.environ)
  • a kubeconfig is available at the default location

# kubernetes.config.kube_config.load_kube_config sets KUBE_CONFIG_DEFAULT_LOCATION during module import.
# If `KUBECONFIG` environment variable is set via code, the `KUBE_CONFIG_DEFAULT_LOCATION` will be None since
# is populated during import which comes before setting the variable in code.
config_file = config_file or os.environ.get("KUBECONFIG", "~/.kube/config")
# If `KUBECONFIG` environment variable is set via code, the `KUBE_CONFIG_DEFAULT_LOCATION` will not be updated
# to reflect the new value since it's set to at import-time.

# Ref: https://github.com/kubernetes-client/python/blob/v26.1.0/kubernetes/base/config/__init__.py
if not config_file:
config_file = os.environ.get("KUBECONFIG", "~/.kube/config")
LOGGER.info("Trying to get client via new_client_from_config config_file=%s", config_file)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use f-string and not %s

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the correct way to pass argument to logging statements: formatting of message arguments is deferred until it cannot be avoided.

https://docs.python.org/3/howto/logging.html#optimization
https://docs.python.org/3/howto/logging.html#logging-variable-data


_client = kubernetes.config.new_client_from_config(
config_file=config_file,
context=context,
client_configuration=client_configuration,
persist_config=persist_config,
)
elif all(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason why not query os.environ with the needed keys?

Copy link
Author

@dtrifiro dtrifiro Oct 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is supposed to be a minimal check to assess whether this is running in a k8s environment, where KUBERNETES_SERVICE_PORT and KUBERNETES_SERVICE_HOST are set.

If these are set, there's a fairly good chance an incluster configuration is available, so we call load_incluster_config which will take care of reading/validating the configuration from env vars and files, see https://github.com/kubernetes-client/python/blob/a486091a91fdd0d0d389bc9538533118ee6fa3c8/kubernetes/base/config/incluster_config.py#L59-L85

var in os.environ
for var in (
kubernetes.config.incluster_config.SERVICE_HOST_ENV_NAME,
kubernetes.config.incluster_config.SERVICE_PORT_ENV_NAME,
)
):
LOGGER.info("Trying to get client via incluster_config")

# Ref: https://github.com/kubernetes-client/python/blob/v26.1.0/kubernetes/base/config/incluster_config.py
kubernetes.config.load_incluster_config(client_configuration)
_client = kubernetes.client.ApiClient(client_configuration)

kubernetes.client.Configuration.set_default(default=client_configuration)

try:
return kubernetes.dynamic.DynamicClient(client=_client)
except MaxRetryError:
# Ref: https://github.com/kubernetes-client/python/blob/v26.1.0/kubernetes/base/config/incluster_config.py
LOGGER.info("Trying to get client via incluster_config")
return kubernetes.dynamic.DynamicClient(
client=kubernetes.config.incluster_config.load_incluster_config(
client_configuration=client_configuration, try_refresh_token=try_refresh_token
),
)
return kubernetes.dynamic.DynamicClient(client=_client)


def sub_resource_level(current_class: Any, owner_class: Any, parent_class: Any) -> str | None:
Expand Down