Merge branch 'master' into add-missing-configs

Author: tejaschumbalkar

src/sagemaker/remote_function/core/serialization.py

@@ -141,7 +141,12 @@ def deserialize(s3_uri: str, bytes_to_deserialize: bytes) -> Any:
             return cloudpickle.loads(bytes_to_deserialize)
         except Exception as e:
             raise DeserializationError(
-                "Error when deserializing bytes downloaded from {}: {}".format(s3_uri, repr(e))
+                "Error when deserializing bytes downloaded from {}: {}. "
+                "NOTE: this may be caused by inconsistent sagemaker python sdk versions "
+                "where remote function runs versus the one used on client side. "
+                "If the sagemaker versions do not match, a warning message would "
+                "be logged starting with 'Inconsistent sagemaker versions found'. "
+                "Please check it to validate.".format(s3_uri, repr(e))
             ) from e
 
 

src/sagemaker/remote_function/job.py

@@ -786,6 +786,12 @@ def compile(
         container_args.extend(
             ["--client_python_version", RuntimeEnvironmentManager()._current_python_version()]
         )
+        container_args.extend(
+            [
+                "--client_sagemaker_pysdk_version",
+                RuntimeEnvironmentManager()._current_sagemaker_pysdk_version(),
+            ]
+        )
         container_args.extend(
             [
                 "--dependency_settings",

src/sagemaker/remote_function/runtime_environment/bootstrap_runtime_environment.py

@@ -56,6 +56,7 @@ def main(sys_args=None):
     try:
         args = _parse_args(sys_args)
         client_python_version = args.client_python_version
+        client_sagemaker_pysdk_version = args.client_sagemaker_pysdk_version
         job_conda_env = args.job_conda_env
         pipeline_execution_id = args.pipeline_execution_id
         dependency_settings = _DependencySettings.from_string(args.dependency_settings)
@@ -64,6 +65,9 @@ def main(sys_args=None):
         conda_env = job_conda_env or os.getenv("SAGEMAKER_JOB_CONDA_ENV")
 
         RuntimeEnvironmentManager()._validate_python_version(client_python_version, conda_env)
+        RuntimeEnvironmentManager()._validate_sagemaker_pysdk_version(
+            client_sagemaker_pysdk_version
+        )
 
         user = getpass.getuser()
         if user != "root":
@@ -274,6 +278,7 @@ def _parse_args(sys_args):
     parser = argparse.ArgumentParser()
     parser.add_argument("--job_conda_env", type=str)
     parser.add_argument("--client_python_version", type=str)
+    parser.add_argument("--client_sagemaker_pysdk_version", type=str, default=None)
     parser.add_argument("--pipeline_execution_id", type=str)
     parser.add_argument("--dependency_settings", type=str)
     parser.add_argument("--func_step_s3_dir", type=str)

src/sagemaker/remote_function/runtime_environment/runtime_environment_manager.py

@@ -24,6 +24,8 @@
 import dataclasses
 import json
 
+import sagemaker
+
 
 class _UTCFormatter(logging.Formatter):
     """Class that overrides the default local time provider in log formatter."""
@@ -326,6 +328,11 @@ def _current_python_version(self):
 
         return f"{sys.version_info.major}.{sys.version_info.minor}".strip()
 
+    def _current_sagemaker_pysdk_version(self):
+        """Returns the current sagemaker python sdk version where program is running"""
+
+        return sagemaker.__version__
+
     def _validate_python_version(self, client_python_version: str, conda_env: str = None):
         """Validate the python version
 
@@ -344,6 +351,29 @@ def _validate_python_version(self, client_python_version: str, conda_env: str =
                 f"is same as the local python version."
             )
 
+    def _validate_sagemaker_pysdk_version(self, client_sagemaker_pysdk_version):
+        """Validate the sagemaker python sdk version
+
+        Validates if the sagemaker python sdk version where remote function runs
+        matches the one used on client side.
+        Otherwise, log a warning to call out that unexpected behaviors
+        may occur in this case.
+        """
+        job_sagemaker_pysdk_version = self._current_sagemaker_pysdk_version()
+        if (
+            client_sagemaker_pysdk_version
+            and client_sagemaker_pysdk_version != job_sagemaker_pysdk_version
+        ):
+            logger.warning(
+                "Inconsistent sagemaker versions found: "
+                "sagemaker pysdk version found in the container is "
+                "'%s' which does not match the '%s' on the local client. "
+                "Please make sure that the python version used in the training container "
+                "is the same as the local python version in case of unexpected behaviors.",
+                job_sagemaker_pysdk_version,
+                client_sagemaker_pysdk_version,
+            )
+
 
 def _run_and_get_output_shell_cmd(cmd: str) -> str:
     """Run and return the output of the given shell command"""

src/sagemaker/transformer.py

@@ -337,6 +337,7 @@ def transform_with_monitoring(
         wait: bool = True,
         pipeline_name: str = None,
         role: str = None,
+        fail_on_violation: bool = True,
     ):
         """Runs a transform job with monitoring job.
 
@@ -352,7 +353,6 @@ def transform_with_monitoring(
             ]): the monitoring configuration used for run model monitoring.
             monitoring_resource_config (`sagemaker.workflow.check_job_config.CheckJobConfig`):
                 the check job (processing job) cluster resource configuration.
-            transform_step_args (_JobStepArguments): the transform step transform arguments.
             data (str): Input data location in S3 for the transform job
             data_type (str): What the S3 location defines (default: 'S3Prefix').
                 Valid values:
@@ -400,8 +400,6 @@ def transform_with_monitoring(
             monitor_before_transform (bgool): If to run data quality
                 or model explainability monitoring type,
                 a true value of this flag indicates running the check step before the transform job.
-            fail_on_violation (Union[bool, PipelineVariable]): A opt-out flag to not to fail the
-                check step when a violation is detected.
             supplied_baseline_statistics (Union[str, PipelineVariable]): The S3 path
                 to the supplied statistics object representing the statistics JSON file
                 which will be used for drift to check (default: None).
@@ -411,6 +409,8 @@ def transform_with_monitoring(
             wait (bool): To determine if needed to wait for the pipeline execution to complete
             pipeline_name (str): The name of the Pipeline for the monitoring and transfrom step
             role (str): Execution role
+            fail_on_violation (Union[bool, PipelineVariable]): A opt-out flag to not to fail the
+                check step when a violation is detected.
         """
 
         transformer = self
@@ -454,6 +454,7 @@ def transform_with_monitoring(
             monitor_before_transform=monitor_before_transform,
             supplied_baseline_constraints=supplied_baseline_constraints,
             supplied_baseline_statistics=supplied_baseline_statistics,
+            fail_on_violation=fail_on_violation,
         )
 
         pipeline_name = (

tests/integ/test_transformer.py

@@ -709,3 +709,67 @@ def test_transformer_and_monitoring_job(
         assert execution_step["StepStatus"] == "Succeeded"
 
     xgb_model.delete_model()
+
+
+def test_transformer_and_monitoring_job_to_pass_with_no_failure_in_violation(
+    pipeline_session,
+    sagemaker_session,
+    role,
+    pipeline_name,
+    check_job_config,
+    data_bias_check_config,
+):
+    xgb_model_data_s3 = pipeline_session.upload_data(
+        path=os.path.join(os.path.join(DATA_DIR, "xgboost_abalone"), "xgb_model.tar.gz"),
+        key_prefix="integ-test-data/xgboost/model",
+    )
+    data_bias_supplied_baseline_constraints = Constraints.from_file_path(
+        constraints_file_path=os.path.join(
+            DATA_DIR, "pipeline/clarify_check_step/data_bias/bad_cases/analysis.json"
+        ),
+        sagemaker_session=sagemaker_session,
+    ).file_s3_uri
+
+    xgb_model = XGBoostModel(
+        model_data=xgb_model_data_s3,
+        framework_version="1.3-1",
+        role=role,
+        sagemaker_session=sagemaker_session,
+        entry_point=os.path.join(os.path.join(DATA_DIR, "xgboost_abalone"), "inference.py"),
+        enable_network_isolation=True,
+    )
+
+    xgb_model.deploy(_INSTANCE_COUNT, _INSTANCE_TYPE)
+
+    transform_output = f"s3://{sagemaker_session.default_bucket()}/{pipeline_name}Transform"
+    transformer = Transformer(
+        model_name=xgb_model.name,
+        strategy="SingleRecord",
+        instance_type="ml.m5.xlarge",
+        instance_count=1,
+        output_path=transform_output,
+        sagemaker_session=pipeline_session,
+    )
+
+    transform_input = pipeline_session.upload_data(
+        path=os.path.join(DATA_DIR, "xgboost_abalone", "abalone"),
+        key_prefix="integ-test-data/xgboost_abalone/abalone",
+    )
+
+    execution = transformer.transform_with_monitoring(
+        monitoring_config=data_bias_check_config,
+        monitoring_resource_config=check_job_config,
+        data=transform_input,
+        content_type="text/libsvm",
+        supplied_baseline_constraints=data_bias_supplied_baseline_constraints,
+        role=role,
+        fail_on_violation=False,
+    )
+
+    execution_steps = execution.list_steps()
+    assert len(execution_steps) == 2
+
+    for execution_step in execution_steps:
+        assert execution_step["StepStatus"] == "Succeeded"
+
+    xgb_model.delete_model()

tests/unit/sagemaker/remote_function/core/test_serialization.py

@@ -198,7 +198,8 @@ def square(x):
     with pytest.raises(
         DeserializationError,
         match=rf"Error when deserializing bytes downloaded from {s3_uri}/payload.pkl: "
-        + r"RuntimeError\('some failure when loads'\)",
+        + r"RuntimeError\('some failure when loads'\). "
+        + r"NOTE: this may be caused by inconsistent sagemaker python sdk versions",
     ):
         deserialize_func_from_s3(sagemaker_session=Mock(), s3_uri=s3_uri, hmac_key=HMAC_KEY)
 
@@ -397,7 +398,8 @@ def __init__(self, x):
     with pytest.raises(
         DeserializationError,
         match=rf"Error when deserializing bytes downloaded from {s3_uri}/payload.pkl: "
-        + r"RuntimeError\('some failure when loads'\)",
+        + r"RuntimeError\('some failure when loads'\). "
+        + r"NOTE: this may be caused by inconsistent sagemaker python sdk versions",
     ):
         deserialize_obj_from_s3(sagemaker_session=Mock(), s3_uri=s3_uri, hmac_key=HMAC_KEY)