From 7f01b40aab71e355eb68983f873e4d0db2700bc2 Mon Sep 17 00:00:00 2001 From: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> Date: Tue, 23 Jun 2026 08:37:32 -0400 Subject: [PATCH 1/3] fix(user_tools): replace deprecated prop containers Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> --- .../cloud_api/dataproc.py | 5 +- .../common/cluster_inference.py | 4 +- .../test_deprecated_prop_container.py | 150 ++++++++++++++++++ 3 files changed, 155 insertions(+), 4 deletions(-) create mode 100644 user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py diff --git a/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py b/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py index 78fe0361c..610a20a16 100644 --- a/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py +++ b/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py @@ -20,6 +20,7 @@ from typing import Any, List, Union, Optional from spark_rapids_tools import CspEnv +from spark_rapids_tools.utils import AbstractPropContainer from spark_rapids_pytools.cloud_api.dataproc_job import DataprocLocalRapidsJob from spark_rapids_pytools.cloud_api.gstorage import GStorageDriver from spark_rapids_pytools.cloud_api.sp_types import PlatformBase, CMDDriverBase, \ @@ -505,7 +506,7 @@ def _init_nodes(self): for worker_node in worker_nodes_from_conf: worker_props = { 'name': worker_node, - 'props': JSONPropertiesContainer(prop_arg=raw_worker_prop, file_load=False), + 'props': AbstractPropContainer(props=raw_worker_prop), # set the node zone based on the wrapper defined zone 'zone': self.zone } @@ -516,7 +517,7 @@ def _init_nodes(self): raw_master_props = self.props.get_value('config', 'masterConfig') master_props = { 'name': master_nodes_from_conf[0], - 'props': JSONPropertiesContainer(prop_arg=raw_master_props, file_load=False), + 'props': AbstractPropContainer(props=raw_master_props), # set the node zone based on the wrapper defined zone 'zone': self.zone } diff --git a/user_tools/src/spark_rapids_pytools/common/cluster_inference.py b/user_tools/src/spark_rapids_pytools/common/cluster_inference.py index 8f5f2a2d7..00df705ba 100644 --- a/user_tools/src/spark_rapids_pytools/common/cluster_inference.py +++ b/user_tools/src/spark_rapids_pytools/common/cluster_inference.py @@ -22,9 +22,9 @@ import pandas as pd from spark_rapids_pytools.cloud_api.sp_types import PlatformBase, ClusterBase -from spark_rapids_pytools.common.prop_manager import JSONPropertiesContainer from spark_rapids_pytools.common.utilities import ToolLogging from spark_rapids_tools import CspEnv +from spark_rapids_tools.utils import AbstractPropContainer class ClusterType(Enum): @@ -145,7 +145,7 @@ def infer_cluster(self, cluster_info_df: pd.DataFrame) -> Optional[ClusterBase]: cluster_conf = self.platform.generate_cluster_configuration(cluster_template_args) if cluster_conf is None: return None - cluster_props_new = JSONPropertiesContainer(cluster_conf, file_load=False) + cluster_props_new = AbstractPropContainer(props=cluster_conf) return self.platform.load_cluster_by_prop(cluster_props_new, is_inferred=True) except Exception as e: # pylint: disable=broad-except self.logger.error('Error while inferring cluster: %s', str(e)) diff --git a/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py b/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py new file mode 100644 index 000000000..b99be91d4 --- /dev/null +++ b/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py @@ -0,0 +1,150 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. +# +# 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. + +"""Tests for deprecated property container replacements.""" + +# pylint: disable=protected-access,too-few-public-methods,wrong-import-position + +import logging +import warnings + +import pandas as pd + + +DEPRECATED_PROP_CONTAINER_MESSAGE = 'Deprecated: use AbstractPropContainer instead' + + +with warnings.catch_warnings(): + warnings.filterwarnings('ignore', message=DEPRECATED_PROP_CONTAINER_MESSAGE, category=DeprecationWarning) + from spark_rapids_pytools.cloud_api.dataproc import DataprocCluster, DataprocNode + from spark_rapids_pytools.cloud_api.sp_types import SparkNodeType + from spark_rapids_pytools.common.cluster_inference import ClusterInference + from spark_rapids_tools import CspEnv + from spark_rapids_tools.utils import AbstractPropContainer + + +class _InferencePlatform: + """Minimal platform implementation for cluster inference tests.""" + + def __init__(self): + self.loaded_cluster_prop = None + self.loaded_is_inferred = None + + @staticmethod + def get_platform_name(): + return CspEnv.DATAPROC + + @staticmethod + def generate_cluster_configuration(render_args): + return { + 'cluster_id': 'inferred-cluster', + 'config': { + 'workerConfig': { + 'numInstances': render_args['NUM_WORKER_NODES'], + 'machineType': render_args['WORKER_NODE_TYPE'].strip('"') + }, + 'masterConfig': { + 'machineType': render_args['DRIVER_NODE_TYPE'].strip('"') + } + } + } + + def load_cluster_by_prop(self, cluster_prop, is_inferred=False): + self.loaded_cluster_prop = cluster_prop + self.loaded_is_inferred = is_inferred + return cluster_prop + + +class _DataprocCli: + """Minimal Dataproc CLI implementation for cluster initialization tests.""" + + def __init__(self): + self.env_vars = {'zone': 'us-central1-a'} + self.logger = logging.getLogger(__name__) + + def get_region(self): + return 'us-central1' + + def get_zone(self): + return self.env_vars['zone'] + + +class _DataprocPlatform: + """Minimal Dataproc platform implementation for cluster initialization tests.""" + + def __init__(self): + self.cli = _DataprocCli() + + +def _deprecated_prop_container_warnings(caught_warnings): + return [ + warning for warning in caught_warnings + if issubclass(warning.category, DeprecationWarning) + and DEPRECATED_PROP_CONTAINER_MESSAGE in str(warning.message) + ] + + +def test_cluster_inference_uses_abstract_prop_container(): + platform = _InferencePlatform() + cluster_info_df = pd.DataFrame([{ + 'App ID': 'app-1', + 'Num Worker Nodes': 2, + 'Cores Per Executor': 4, + 'Num Executors Per Node': 1, + 'Driver Node Type': 'n1-standard-4', + 'Worker Node Type': 'n1-standard-8' + }]) + + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter('always', DeprecationWarning) + inferred_cluster = ClusterInference(platform=platform).infer_cluster(cluster_info_df) + + assert inferred_cluster is platform.loaded_cluster_prop + assert platform.loaded_is_inferred is True + assert isinstance(platform.loaded_cluster_prop, AbstractPropContainer) + assert not _deprecated_prop_container_warnings(caught_warnings) + + +def test_dataproc_init_nodes_uses_abstract_prop_container(monkeypatch): + def skip_fetching_hw_info(self, cli): + del self, cli + + monkeypatch.setattr(DataprocNode, 'fetch_and_set_hw_info', skip_fetching_hw_info) + cluster = DataprocCluster(_DataprocPlatform(), is_inferred=True) + cluster.zone = 'us-central1-a' + cluster.props = AbstractPropContainer(props={ + 'config': { + 'masterConfig': { + 'instanceNames': ['master-0'], + 'machineType': 'n1-standard-4' + }, + 'workerConfig': { + 'numInstances': 1, + 'instanceNames': ['worker-0'], + 'machineType': 'n1-standard-8' + } + } + }) + + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter('always', DeprecationWarning) + cluster._init_nodes() + + master_node = cluster.nodes[SparkNodeType.MASTER] + worker_node = cluster.nodes[SparkNodeType.WORKER][0] + assert isinstance(master_node.props, AbstractPropContainer) + assert isinstance(worker_node.props, AbstractPropContainer) + assert master_node.instance_type == 'n1-standard-4' + assert worker_node.instance_type == 'n1-standard-8' + assert not _deprecated_prop_container_warnings(caught_warnings) From 0d35081ad2b8b2e69058c4c31b7fe8a40894cf11 Mon Sep 17 00:00:00 2001 From: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> Date: Tue, 23 Jun 2026 10:11:24 -0400 Subject: [PATCH 2/3] fix(user_tools): replace remaining Dataproc prop containers Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> --- .../cloud_api/dataproc.py | 24 +++-- .../test_deprecated_prop_container.py | 101 ++++++++++++++++++ 2 files changed, 118 insertions(+), 7 deletions(-) diff --git a/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py b/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py index 610a20a16..25f3182d3 100644 --- a/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py +++ b/user_tools/src/spark_rapids_pytools/cloud_api/dataproc.py @@ -15,6 +15,7 @@ """Implementation specific to Dataproc""" +import json from collections import defaultdict from dataclasses import dataclass, field from typing import Any, List, Union, Optional @@ -26,13 +27,24 @@ from spark_rapids_pytools.cloud_api.sp_types import PlatformBase, CMDDriverBase, \ ClusterBase, ClusterNode, SysInfo, GpuHWInfo, SparkNodeType, ClusterState, GpuDevice, \ NodeHWInfo, ClusterGetAccessor -from spark_rapids_pytools.common.prop_manager import JSONPropertiesContainer, is_valid_gpu_device +from spark_rapids_pytools.common.prop_manager import is_valid_gpu_device from spark_rapids_pytools.common.sys_storage import FSUtil from spark_rapids_pytools.common.utilities import Utils from spark_rapids_pytools.pricing.dataproc_pricing import DataprocPriceProvider from spark_rapids_pytools.pricing.price_provider import SavingsEstimator +def _load_json_props(prop_arg: Any) -> Any: + if not isinstance(prop_arg, str): + return prop_arg + try: + return json.loads(prop_arg) + except json.JSONDecodeError as exc: + raise RuntimeError('Incorrect format of JSON File') from exc + except TypeError as exc: + raise RuntimeError('Incorrect Type of JSON content') from exc + + @dataclass class DataprocPlatform(PlatformBase): """ @@ -114,11 +126,9 @@ def create_saving_estimator(self, target_cost: float = None, source_cost: float = None): raw_pricing_config = self.configs.get_value_silent('pricing') + pricing_config: Optional[AbstractPropContainer] = None if raw_pricing_config: - pricing_config = JSONPropertiesContainer(prop_arg=raw_pricing_config, - file_load=False) - else: - pricing_config: JSONPropertiesContainer = None + pricing_config = AbstractPropContainer(props=_load_json_props(raw_pricing_config)) pricing_provider = DataprocPriceProvider(region=self.cli.get_region(), pricing_configs={'gcloud': pricing_config}) saving_estimator = DataprocSavingsEstimator(price_provider=pricing_provider, @@ -320,7 +330,7 @@ def extract_gpu_name(gpu_description: str) -> str: return gpu_name.upper() processed_instance_descriptions = {} - raw_instances_descriptions = JSONPropertiesContainer(prop_arg=instance_descriptions, file_load=False) + raw_instances_descriptions = AbstractPropContainer(props=_load_json_props(instance_descriptions)) for instance in raw_instances_descriptions.props: instance_content = {} instance_content['VCpuCount'] = int(instance.get('guestCpus', -1)) @@ -528,7 +538,7 @@ def _init_nodes(self): SparkNodeType.MASTER: master_node } - def _set_zone_from_props(self, prop_container: JSONPropertiesContainer): + def _set_zone_from_props(self, prop_container: AbstractPropContainer): """ Extracts the 'zoneUri' from the properties container and updates the environment variable dictionary. """ diff --git a/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py b/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py index b99be91d4..7f08fd8aa 100644 --- a/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py +++ b/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py @@ -16,6 +16,7 @@ # pylint: disable=protected-access,too-few-public-methods,wrong-import-position +import json import logging import warnings @@ -27,6 +28,7 @@ with warnings.catch_warnings(): warnings.filterwarnings('ignore', message=DEPRECATED_PROP_CONTAINER_MESSAGE, category=DeprecationWarning) + from spark_rapids_pytools.cloud_api import dataproc as dataproc_mod from spark_rapids_pytools.cloud_api.dataproc import DataprocCluster, DataprocNode from spark_rapids_pytools.cloud_api.sp_types import SparkNodeType from spark_rapids_pytools.common.cluster_inference import ClusterInference @@ -87,6 +89,26 @@ def __init__(self): self.cli = _DataprocCli() +class _DataprocConfigs: + """Minimal Dataproc configs implementation for pricing tests.""" + + def __init__(self, pricing): + self.pricing = pricing + + def get_value_silent(self, *keys): + if keys == ('pricing',): + return self.pricing + return None + + +class _DataprocPricingPlatform(_DataprocPlatform): + """Minimal Dataproc platform implementation for savings estimator tests.""" + + def __init__(self, pricing): + super().__init__() + self.configs = _DataprocConfigs(pricing) + + def _deprecated_prop_container_warnings(caught_warnings): return [ warning for warning in caught_warnings @@ -148,3 +170,82 @@ def skip_fetching_hw_info(self, cli): assert master_node.instance_type == 'n1-standard-4' assert worker_node.instance_type == 'n1-standard-8' assert not _deprecated_prop_container_warnings(caught_warnings) + + +def test_dataproc_pricing_config_uses_abstract_prop_container(monkeypatch): + pricing_props = { + 'catalog': { + 'onlineResources': [{ + 'resourceKey': 'gcloud-catalog', + 'localFile': 'gcloud-catalog.json', + 'onlineURL': 'https://example.com/gcloud-catalog.json' + }] + } + } + + class RecordingDataprocPriceProvider: + def __init__(self, region, pricing_configs): + self.region = region + self.pricing_configs = pricing_configs + + class RecordingDataprocSavingsEstimator: + def __init__(self, price_provider, reshaped_cluster, source_cluster, + target_cost=None, source_cost=None): + self.price_provider = price_provider + self.reshaped_cluster = reshaped_cluster + self.source_cluster = source_cluster + self.target_cost = target_cost + self.source_cost = source_cost + + monkeypatch.setattr(dataproc_mod, 'DataprocPriceProvider', RecordingDataprocPriceProvider) + monkeypatch.setattr(dataproc_mod, 'DataprocSavingsEstimator', RecordingDataprocSavingsEstimator) + platform = _DataprocPricingPlatform(json.dumps(pricing_props)) + + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter('always', DeprecationWarning) + estimator = dataproc_mod.DataprocPlatform.create_saving_estimator( + platform, + source_cluster=object(), + reshaped_cluster=object() + ) + + pricing_config = estimator.price_provider.pricing_configs['gcloud'] + assert estimator.price_provider.region == 'us-central1' + assert isinstance(pricing_config, AbstractPropContainer) + assert pricing_config.props == pricing_props + assert not _deprecated_prop_container_warnings(caught_warnings) + + +def test_dataproc_instance_description_uses_abstract_prop_container(monkeypatch): + instance_descriptions = [{ + 'name': 'a2-highgpu-1g', + 'guestCpus': 12, + 'memoryMb': 87296, + 'accelerators': [{ + 'guestAcceleratorType': 'nvidia-tesla-a100', + 'guestAcceleratorCount': 1 + }] + }] + captured_props = [] + + class RecordingPropContainer(AbstractPropContainer): + def __init__(self, **kwargs): + super().__init__(**kwargs) + captured_props.append(self.props) + + monkeypatch.setattr(dataproc_mod, 'AbstractPropContainer', RecordingPropContainer) + + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter('always', DeprecationWarning) + processed_instances = dataproc_mod.DataprocCMDDriver._process_instance_description( + None, + json.dumps(instance_descriptions) + ) + + assert captured_props == [instance_descriptions] + assert processed_instances['a2-highgpu-1g'] == { + 'VCpuCount': 12, + 'MemoryInMB': 87296, + 'GpuInfo': [{'Name': 'A100', 'Count': [1]}] + } + assert not _deprecated_prop_container_warnings(caught_warnings) From 1a91b804d5b5eb7881e2a7fadb644aa0a17e75d9 Mon Sep 17 00:00:00 2001 From: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> Date: Wed, 26 Aug 2026 08:12:28 +0800 Subject: [PATCH 3/3] fix(user_tools): parse inferred cluster JSON Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> --- .../src/spark_rapids_pytools/common/cluster_inference.py | 3 ++- .../spark_rapids_tools_ut/test_deprecated_prop_container.py | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/user_tools/src/spark_rapids_pytools/common/cluster_inference.py b/user_tools/src/spark_rapids_pytools/common/cluster_inference.py index 00df705ba..5b67c49b9 100644 --- a/user_tools/src/spark_rapids_pytools/common/cluster_inference.py +++ b/user_tools/src/spark_rapids_pytools/common/cluster_inference.py @@ -14,6 +14,7 @@ """This module provides functionality for cluster inference""" +import json from dataclasses import dataclass, field from enum import Enum from typing import Optional @@ -145,7 +146,7 @@ def infer_cluster(self, cluster_info_df: pd.DataFrame) -> Optional[ClusterBase]: cluster_conf = self.platform.generate_cluster_configuration(cluster_template_args) if cluster_conf is None: return None - cluster_props_new = AbstractPropContainer(props=cluster_conf) + cluster_props_new = AbstractPropContainer(props=json.loads(cluster_conf)) return self.platform.load_cluster_by_prop(cluster_props_new, is_inferred=True) except Exception as e: # pylint: disable=broad-except self.logger.error('Error while inferring cluster: %s', str(e)) diff --git a/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py b/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py index 7f08fd8aa..0a4f223cc 100644 --- a/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py +++ b/user_tools/tests/spark_rapids_tools_ut/test_deprecated_prop_container.py @@ -49,7 +49,7 @@ def get_platform_name(): @staticmethod def generate_cluster_configuration(render_args): - return { + return json.dumps({ 'cluster_id': 'inferred-cluster', 'config': { 'workerConfig': { @@ -60,7 +60,7 @@ def generate_cluster_configuration(render_args): 'machineType': render_args['DRIVER_NODE_TYPE'].strip('"') } } - } + }) def load_cluster_by_prop(self, cluster_prop, is_inferred=False): self.loaded_cluster_prop = cluster_prop @@ -135,6 +135,8 @@ def test_cluster_inference_uses_abstract_prop_container(): assert inferred_cluster is platform.loaded_cluster_prop assert platform.loaded_is_inferred is True assert isinstance(platform.loaded_cluster_prop, AbstractPropContainer) + assert platform.loaded_cluster_prop.props['cluster_id'] == 'inferred-cluster' + assert platform.loaded_cluster_prop.get_value('config', 'workerConfig', 'numInstances') == 2 assert not _deprecated_prop_container_warnings(caught_warnings)