From e345e29989b2428cc571a95cc78e460b08571170 Mon Sep 17 00:00:00 2001 From: Paul Mathon Date: Mon, 26 Jan 2026 17:17:57 +0000 Subject: [PATCH 1/7] New PEX uploads should be based on pex_hash property Custer Pack is relying on the code_hash property of the PEX-INFO metadata file to determine whether it should upload a new env to the /user/{USER}/envs folder on HDFS. According PEX' PyDoc (https://github.com/pex-tool/pex/blob/main/pex/pex_info.py): - pex_hash: sha1 hash of all names/code and distributions in the pex - code_hash: sha1 hash of all names/code in the archive If dependencies change, the PEX changes as well, we should upload a new PEX, to do so we should rely on the pex_hash property rather that than the code_hash. --- cluster_pack/uploader.py | 2 +- tests/test_uploader.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cluster_pack/uploader.py b/cluster_pack/uploader.py index bedbcc1..b7502cc 100644 --- a/cluster_pack/uploader.py +++ b/cluster_pack/uploader.py @@ -271,7 +271,7 @@ def _upload_pex_file( info_to_upload = PexInfo.from_pex(pex_file_or_dir) if ( not force_upload - and info_from_storage.code_hash == info_to_upload.code_hash + and info_from_storage.pex_hash == info_to_upload.pex_hash ): _logger.info( f"skip upload of current {pex_file_or_dir}" diff --git a/tests/test_uploader.py b/tests/test_uploader.py index f6e99c4..4ca8496 100644 --- a/tests/test_uploader.py +++ b/tests/test_uploader.py @@ -327,9 +327,9 @@ def test_upload_env_in_a_pex(): def _from_pex(arg): if arg == f"{home_path}/myapp.pex": - return PexInfo({"code_hash": 1}) + return PexInfo({"pex_hash": 1}) else: - return PexInfo({"code_hash": 2}) + return PexInfo({"pex_hash": 2}) mock_pex_info.from_pex.side_effect = _from_pex From 49682099ec9899d1ef8fee543b824429f949da27 Mon Sep 17 00:00:00 2001 From: Alexander Mazurov Date: Mon, 26 Jan 2026 15:35:24 +0100 Subject: [PATCH 2/7] feat(packaging): Add support for custom user via environment Introduce _get_current_user() function to allow overriding the current user via the C_PACK_USER environment variable. This provides more flexibility when generating package paths, especially in containerized or shared environments where the system user might not match the desired user. Replaces direct calls to getpass.getuser() with the new function, ensuring consistent user resolution across the packaging module. --- cluster_pack/packaging.py | 17 +++++++- cluster_pack/skein/skein_launcher.py | 3 +- tests/test_packaging.py | 62 ++++++++++++++++++++++++---- 3 files changed, 72 insertions(+), 10 deletions(-) diff --git a/cluster_pack/packaging.py b/cluster_pack/packaging.py index 3f0d182..e1bbfa7 100644 --- a/cluster_pack/packaging.py +++ b/cluster_pack/packaging.py @@ -53,6 +53,21 @@ def _get_tmp_dir() -> str: return tmp_dir +def _get_current_user() -> str: + """ + Get the current user name. + + First checks the C_PACK_USER environment variable. + If not set or empty, falls back to getpass.getuser(). + + :return: username string + """ + user = os.environ.get("C_PACK_USER", "").strip() + if user: + return user + return getpass.getuser() + + def zip_path( py_dir: str, include_base_name: bool = True, tmp_dir: str = _get_tmp_dir() ) -> str: @@ -373,7 +388,7 @@ def get_non_editable_requirements(executable: str = sys.executable) -> Dict[str, def _build_package_path(name: str, extension: Optional[str]) -> str: - path = f"{get_default_fs()}/user/{getpass.getuser()}/envs/{name}" + path = f"{get_default_fs()}/user/{_get_current_user()}/envs/{name}" if extension is None: return path return f"{path}.{extension}" diff --git a/cluster_pack/skein/skein_launcher.py b/cluster_pack/skein/skein_launcher.py index 1195bcf..a54f8e2 100644 --- a/cluster_pack/skein/skein_launcher.py +++ b/cluster_pack/skein/skein_launcher.py @@ -10,6 +10,7 @@ from cluster_pack.skein import skein_config_builder from cluster_pack import filesystem +from cluster_pack.packaging import _get_current_user logger = logging.getLogger(__name__) @@ -223,7 +224,7 @@ def _submit( spec.acquire_map_reduce_delegation_token = acquire_map_reduce_delegation_token # activate Impersonation only if user to run the job is not the current user (yarn issue) - if user and user != getpass.getuser(): + if user and user != _get_current_user(): spec.user = user if queue: diff --git a/tests/test_packaging.py b/tests/test_packaging.py index c111968..af935ff 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -22,7 +22,8 @@ LARGE_PEX_CMD, resolve_zip_from_pex_dir, check_large_pex, - PexTooLargeError + PexTooLargeError, + _get_current_user ) MODULE_TO_TEST = "cluster_pack.packaging" @@ -44,6 +45,51 @@ def test_get_virtualenv_empty_returns_default(): assert "default" == packaging.get_env_name(VARNAME) +def test_get_current_user_with_env_variable(): + """Test that C_PACK_USER environment variable is used when set.""" + with mock.patch.dict("os.environ"): + # Test when C_PACK_USER is set + os.environ["C_PACK_USER"] = "custom_user" + assert packaging._get_current_user() == "custom_user" + + +def test_get_current_user_with_empty_env_variable(): + """Test that empty C_PACK_USER falls back to getpass.getuser().""" + with contextlib.ExitStack() as stack: + stack.enter_context(mock.patch(f"{MODULE_TO_TEST}.getpass.getuser", return_value="system_user")) + with mock.patch.dict("os.environ", clear=True): + # Test when C_PACK_USER is not set + assert packaging._get_current_user() == "system_user" + + # Test when C_PACK_USER is empty string + os.environ["C_PACK_USER"] = "" + assert packaging._get_current_user() == "system_user" + + # Test when C_PACK_USER is only whitespace + os.environ["C_PACK_USER"] = " " + assert packaging._get_current_user() == "system_user" + + +def test_get_current_user_strips_whitespace(): + """Test that C_PACK_USER whitespace is stripped.""" + with mock.patch.dict("os.environ"): + os.environ["C_PACK_USER"] = " spaced_user " + assert packaging._get_current_user() == "spaced_user" + + +def test_build_package_path_uses_c_pack_user_env(): + """Test that _build_package_path uses C_PACK_USER when set.""" + with contextlib.ExitStack() as stack: + stack.enter_context(mock.patch(f"{MODULE_TO_TEST}.get_default_fs", return_value="hdfs://")) + with mock.patch.dict("os.environ"): + os.environ["C_PACK_USER"] = "env_user" + + result = packaging._build_package_path("myenv", "pex") + expected = "hdfs:///user/env_user/envs/myenv.pex" + + assert result == expected + + def test_get_empty_editable_requirements(): with tempfile.TemporaryDirectory() as tempdir: _create_venv(tempdir) @@ -530,19 +576,19 @@ def test_gen_pyenvs_from_unknown_format(): (False, "dummy/path/exe.pex", True, False, "dummy/path/exe.pex.zip"), (True, "dummy/path/exe.pex", False, False, "dummy/path/exe.pex"), (True, "dummy/path/exe.pex", True, False, "dummy/path/exe.pex.zip"), - (False, None, False, False, f"hdfs:///user/{getpass.getuser()}/envs/venv_exe.pex"), - (False, None, None, False, f"hdfs:///user/{getpass.getuser()}/envs/venv_exe.pex"), + (False, None, False, False, f"hdfs:///user/{_get_current_user()}/envs/venv_exe.pex"), + (False, None, None, False, f"hdfs:///user/{_get_current_user()}/envs/venv_exe.pex"), ( False, None, True, False, - f"hdfs:///user/{getpass.getuser()}/envs/venv_exe.pex.zip", + f"hdfs:///user/{_get_current_user()}/envs/venv_exe.pex.zip", ), - (True, None, False, False, f"hdfs:///user/{getpass.getuser()}/envs/pex_exe.pex"), - (True, None, True, False, f"hdfs:///user/{getpass.getuser()}/envs/pex_exe.pex.zip"), - (True, None, None, False, f"hdfs:///user/{getpass.getuser()}/envs/pex_exe.pex"), - (True, None, None, True, f"hdfs:///user/{getpass.getuser()}/envs/pex_exe.pex.zip"), + (True, None, False, False, f"hdfs:///user/{_get_current_user()}/envs/pex_exe.pex"), + (True, None, True, False, f"hdfs:///user/{_get_current_user()}/envs/pex_exe.pex.zip"), + (True, None, None, False, f"hdfs:///user/{_get_current_user()}/envs/pex_exe.pex"), + (True, None, None, True, f"hdfs:///user/{_get_current_user()}/envs/pex_exe.pex.zip"), ] From 67986d51bf8d94c71150ec039372e25d26f5cb9d Mon Sep 17 00:00:00 2001 From: Alexander Mazurov Date: Mon, 26 Jan 2026 16:50:04 +0100 Subject: [PATCH 3/7] refactor(tests): Improve readability of test context managers Refactor test context managers in packaging tests to use multi-line formatting for better readability and consistency. This change enhances code clarity without altering test logic. --- tests/test_packaging.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_packaging.py b/tests/test_packaging.py index af935ff..5566389 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -56,7 +56,9 @@ def test_get_current_user_with_env_variable(): def test_get_current_user_with_empty_env_variable(): """Test that empty C_PACK_USER falls back to getpass.getuser().""" with contextlib.ExitStack() as stack: - stack.enter_context(mock.patch(f"{MODULE_TO_TEST}.getpass.getuser", return_value="system_user")) + stack.enter_context( + mock.patch(f"{MODULE_TO_TEST}.getpass.getuser", return_value="system_user") + ) with mock.patch.dict("os.environ", clear=True): # Test when C_PACK_USER is not set assert packaging._get_current_user() == "system_user" @@ -80,7 +82,9 @@ def test_get_current_user_strips_whitespace(): def test_build_package_path_uses_c_pack_user_env(): """Test that _build_package_path uses C_PACK_USER when set.""" with contextlib.ExitStack() as stack: - stack.enter_context(mock.patch(f"{MODULE_TO_TEST}.get_default_fs", return_value="hdfs://")) + stack.enter_context( + mock.patch(f"{MODULE_TO_TEST}.get_default_fs", return_value="hdfs://") + ) with mock.patch.dict("os.environ"): os.environ["C_PACK_USER"] = "env_user" From b061227da4e6405396657d41f42952e25ba88261 Mon Sep 17 00:00:00 2001 From: Alexander Mazurov Date: Wed, 28 Jan 2026 17:51:59 +0100 Subject: [PATCH 4/7] feat(config): Add C_PACK_USER feature flag for user override to README --- README.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/README.md b/README.md index 8108941..b5f4943 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,12 @@ $ pip install . cluster-pack supports Python ≥3.9. +## Feature flags +- C_PACK_USER: override the current user for HDFS path generation and Skein impersonation + - When set, this value is used instead of the system user (from `getpass.getuser()`) + - Useful for running jobs as a different user or in environments where the system user doesn't match the HDFS user + - If not set or empty, falls back to the current system user + ## Features - Ships a package with all the dependencies from your current virtual environment From 34d2229b2d3f2417003109a253f56d13be62dc8a Mon Sep 17 00:00:00 2001 From: "q.auge" Date: Tue, 3 Feb 2026 18:40:20 +0100 Subject: [PATCH 5/7] Upgrade spark version in CI Minor version no longer available --- .github/workflows/main.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index a776632..bdf57bf 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -54,7 +54,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - version: [ { python: 3.9.15, spark: 3.5.7 }, { python: 3.11.8, spark: 3.5.7 } ] + version: [ { python: 3.9.15, spark: 3.5.8 }, { python: 3.11.8, spark: 3.5.8 } ] steps: - uses: actions/checkout@v2 From b8350e68b403d7220dc37a9f091b2878fdcf48b7 Mon Sep 17 00:00:00 2001 From: "q.auge" Date: Tue, 3 Feb 2026 19:13:05 +0100 Subject: [PATCH 6/7] Fix imp deprecation --- cluster_pack/packaging.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/cluster_pack/packaging.py b/cluster_pack/packaging.py index e1bbfa7..22f3432 100644 --- a/cluster_pack/packaging.py +++ b/cluster_pack/packaging.py @@ -1,5 +1,5 @@ import getpass -import imp +import importlib.util import json import logging import os @@ -504,9 +504,12 @@ def get_editable_requirements( else: for package_name in package_names: try: - _, path, _ = imp.find_module(package_name) + spec = importlib.util.find_spec(package_name) + if spec is None or spec.origin is None: + raise ModuleNotFoundError(f"No module named '{package_name}'") + path = os.path.dirname(spec.origin) editable_requirements[os.path.basename(path)] = path - except ImportError: + except ModuleNotFoundError: _logger.error( f"Could not import package {package_name}" f" repo exists={os.path.exists(package_name)}" From fced4c969a8db9e48e9937380835756dfd6dda54 Mon Sep 17 00:00:00 2001 From: "q.auge" Date: Tue, 3 Feb 2026 19:18:15 +0100 Subject: [PATCH 7/7] Fix broken test --- tests/test_packaging.py | 22 +++++++--------------- 1 file changed, 7 insertions(+), 15 deletions(-) diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 5566389..54534d0 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -293,18 +293,10 @@ def does_not_raise(): def test_pack_in_pex(): - if sys.version_info.minor == 9: - requirements = [ - "protobuf==3.19.6", - "tensorflow==2.5.2", - "tensorboard==2.10.1", - "pyarrow==6.0.1", - ] - else: - requirements = [ - "tensorflow", - "pyarrow" - ] + requirements = [ + "numpy", + "pyarrow" + ] with tempfile.TemporaryDirectory() as tempdir: packaging.pack_in_pex( requirements, f"{tempdir}/out.pex", pex_inherit_path="false" @@ -317,9 +309,9 @@ def test_pack_in_pex(): f"{tempdir}/out.pex", "-c", ( - """print("Start importing pyarrow and tensorflow..");""" - """import pyarrow; import tensorflow;""" - """print("Successfully imported pyarrow and tensorflow!")""" + """print("Start importing pyarrow and numpy..");""" + """import pyarrow; import numpy;""" + """print("Successfully imported pyarrow and numpy!")""" ), ] )