Merge branch 'master' into fix/pipeline-variable-kms-key

Author: evakravi

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## v2.177.1 (2023-08-14)
+
+### Bug Fixes and Other Changes
+
+ * chore: excessive jumpstart bucket logging
+
 ## v2.177.0 (2023-08-11)
 
 ### Features

VERSION

@@ -1 +1 @@
-2.177.1.dev0
+2.177.2.dev0

src/sagemaker/jumpstart/model.py

@@ -310,13 +310,34 @@ def _is_valid_model_id_hook():
 
         super(JumpStartModel, self).__init__(**model_init_kwargs.to_kwargs_dict())
 
-    def _create_sagemaker_model(self, *args, **kwargs):  # pylint: disable=unused-argument
+    def _create_sagemaker_model(
+        self,
+        instance_type=None,
+        accelerator_type=None,
+        tags=None,
+        serverless_inference_config=None,
+        **kwargs,
+    ):
         """Create a SageMaker Model Entity
 
         Args:
-            args: Positional arguments coming from the caller. This class does not require
-                any so they are ignored.
-
+            instance_type (str): Optional. The EC2 instance type that this Model will be
+                used for, this is only used to determine if the image needs GPU
+                support or not. (Default: None).
+            accelerator_type (str): Optional. Type of Elastic Inference accelerator to
+                attach to an endpoint for model loading and inference, for
+                example, 'ml.eia1.medium'. If not specified, no Elastic
+                Inference accelerator will be attached to the endpoint. (Default: None).
+            tags (List[dict[str, str]]): Optional. The list of tags to add to
+                the model. Example: >>> tags = [{'Key': 'tagname', 'Value':
+                'tagvalue'}] For more information about tags, see
+                https://boto3.amazonaws.com/v1/documentation
+                /api/latest/reference/services/sagemaker.html#SageMaker.Client.add_tags
+                (Default: None).
+            serverless_inference_config (sagemaker.serverless.ServerlessInferenceConfig):
+                Optional. Specifies configuration related to serverless endpoint. Instance type is
+                not provided in serverless inference. So this is used to find image URIs.
+                (Default: None).
             kwargs: Keyword arguments coming from the caller. This class does not require
                 any so they are ignored.
         """
@@ -347,10 +368,16 @@ def _create_sagemaker_model(self, *args, **kwargs):  # pylint: disable=unused-ar
                 container_def,
                 vpc_config=self.vpc_config,
                 enable_network_isolation=self.enable_network_isolation(),
-                tags=kwargs.get("tags"),
+                tags=tags,
             )
         else:
-            super(JumpStartModel, self)._create_sagemaker_model(*args, **kwargs)
+            super(JumpStartModel, self)._create_sagemaker_model(
+                instance_type=instance_type,
+                accelerator_type=accelerator_type,
+                tags=tags,
+                serverless_inference_config=serverless_inference_config,
+                **kwargs,
+            )
 
     def deploy(
         self,

src/sagemaker/model.py

@@ -1375,7 +1375,10 @@ def deploy(
                 self._base_name = "-".join((self._base_name, compiled_model_suffix))
 
         self._create_sagemaker_model(
-            instance_type, accelerator_type, tags, serverless_inference_config
+            instance_type=instance_type,
+            accelerator_type=accelerator_type,
+            tags=tags,
+            serverless_inference_config=serverless_inference_config,
         )
 
         serverless_inference_config_dict = (

src/sagemaker/model_monitor/clarify_model_monitoring.py

@@ -27,6 +27,7 @@
 from sagemaker.session import Session
 from sagemaker.utils import name_from_base
 from sagemaker.clarify import SageMakerClarifyProcessor, ModelPredictedLabelConfig
+from sagemaker.lineage._utils import get_resource_name_from_arn
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -154,6 +155,29 @@ def list_executions(self):
             for execution in executions
         ]
 
+    def get_latest_execution_logs(self, wait=False):
+        """Get the processing job logs for the most recent monitoring execution
+
+        Args:
+            wait (bool): Whether the call should wait until the job completes (default: False).
+
+        Raises:
+            ValueError: If no execution job or processing job for the last execution has run
+
+        Returns: None
+        """
+        monitoring_executions = self.sagemaker_session.list_monitoring_executions(
+            monitoring_schedule_name=self.monitoring_schedule_name
+        )
+        if len(monitoring_executions["MonitoringExecutionSummaries"]) == 0:
+            raise ValueError("No execution jobs were kicked off.")
+        if "ProcessingJobArn" not in monitoring_executions["MonitoringExecutionSummaries"][0]:
+            raise ValueError("Processing Job did not run for the last execution")
+        job_arn = monitoring_executions["MonitoringExecutionSummaries"][0]["ProcessingJobArn"]
+        self.sagemaker_session.logs_for_processing_job(
+            job_name=get_resource_name_from_arn(job_arn), wait=wait
+        )
+
     def _create_baselining_processor(self):
         """Create and return a SageMakerClarifyProcessor object which will run the baselining job.
 

src/sagemaker/model_monitor/model_monitoring.py

@@ -63,6 +63,7 @@
     resolve_value_from_config,
     resolve_class_attribute_from_config,
 )
+from sagemaker.lineage._utils import get_resource_name_from_arn
 
 DEFAULT_REPOSITORY_NAME = "sagemaker-model-monitor-analyzer"
 
@@ -768,6 +769,29 @@ def list_executions(self):
 
         return monitoring_executions
 
+    def get_latest_execution_logs(self, wait=False):
+        """Get the processing job logs for the most recent monitoring execution
+
+        Args:
+            wait (bool): Whether the call should wait until the job completes (default: False).
+
+        Raises:
+            ValueError: If no execution job or processing job for the last execution has run
+
+        Returns: None
+        """
+        monitoring_executions = self.sagemaker_session.list_monitoring_executions(
+            monitoring_schedule_name=self.monitoring_schedule_name
+        )
+        if len(monitoring_executions["MonitoringExecutionSummaries"]) == 0:
+            raise ValueError("No execution jobs were kicked off.")
+        if "ProcessingJobArn" not in monitoring_executions["MonitoringExecutionSummaries"][0]:
+            raise ValueError("Processing Job did not run for the last execution")
+        job_arn = monitoring_executions["MonitoringExecutionSummaries"][0]["ProcessingJobArn"]
+        self.sagemaker_session.logs_for_processing_job(
+            job_name=get_resource_name_from_arn(job_arn), wait=wait
+        )
+
     def update_monitoring_alert(
         self,
         monitoring_alert_name: str,

tests/integ/test_model_monitor.py

@@ -486,6 +486,38 @@ def test_default_monitor_suggest_baseline_and_create_monitoring_schedule_with_cu
     assert len(summary["MonitoringScheduleSummaries"]) > 0
 
 
+def test_default_monitor_display_logs_errors(sagemaker_session):
+    my_default_monitor = DefaultModelMonitor(role=ROLE, sagemaker_session=sagemaker_session)
+
+    data_captured_destination_s3_uri = os.path.join(
+        "s3://",
+        sagemaker_session.default_bucket(),
+        "sagemaker-serving-batch-transform",
+        str(uuid.uuid4()),
+    )
+
+    batch_transform_input = BatchTransformInput(
+        data_captured_destination_s3_uri=data_captured_destination_s3_uri,
+        destination="/opt/ml/processing/output",
+        dataset_format=MonitoringDatasetFormat.csv(header=False),
+    )
+
+    my_default_monitor.create_monitoring_schedule(
+        batch_transform_input=batch_transform_input,
+        schedule_cron_expression=CronExpressionGenerator.hourly(),
+    )
+
+    _wait_for_schedule_changes_to_apply(monitor=my_default_monitor)
+
+    try:
+        my_default_monitor.get_latest_execution_logs(wait=False)
+    except ValueError as ve:
+        assert "No execution jobs were kicked off." in str(ve)
+
+    my_default_monitor.stop_monitoring_schedule()
+    my_default_monitor.delete_monitoring_schedule()
+
+
 @pytest.mark.skipif(
     tests.integ.test_region() in tests.integ.NO_MODEL_MONITORING_REGIONS,
     reason="ModelMonitoring is not yet supported in this region.",
@@ -1643,6 +1675,7 @@ def test_byoc_monitor_attach_followed_by_baseline_and_update_monitoring_schedule
     output_kms_key,
     updated_volume_kms_key,
     updated_output_kms_key,
+    capfd,
 ):
     baseline_dataset = os.path.join(DATA_DIR, "monitor/baseline_dataset.csv")
 
@@ -1771,6 +1804,10 @@ def test_byoc_monitor_attach_followed_by_baseline_and_update_monitoring_schedule
 
     _wait_for_schedule_changes_to_apply(monitor=my_attached_monitor)
 
+    _check_processing_logs_generated(
+        monitor=my_attached_monitor, schedule_description=schedule_description, capfd=capfd
+    )
+
     my_attached_monitor.stop_monitoring_schedule()
 
     _wait_for_schedule_changes_to_apply(monitor=my_attached_monitor)
@@ -1877,6 +1914,13 @@ def test_default_monitor_monitoring_alerts(sagemaker_session, predictor):
     my_default_monitor.delete_monitoring_schedule()
 
 
+def _check_processing_logs_generated(monitor, schedule_description, capfd):
+    monitor.get_latest_execution_logs(wait=False)
+    out, _ = capfd.readouterr()
+    assert len(out) > 0
+    assert schedule_description.get("LastMonitoringExecutionSummary")["ProcessingJobArn"] in out
+
+
 def _wait_for_schedule_changes_to_apply(monitor):
     """Waits for the monitor to no longer be in the 'Pending' state. Updates take under a minute
     to apply.

tests/unit/sagemaker/jumpstart/model/test_model.py

@@ -589,7 +589,10 @@ def test_jumpstart_model_package_arn(
 
         model = JumpStartModel(model_id=model_id)
 
-        model.deploy()
+        tag = {"Key": "foo", "Value": "bar"}
+        tags = [tag]
+
+        model.deploy(tags=tags)
 
         self.assertEqual(
             mock_session.return_value.create_model.call_args[0][2],
@@ -599,6 +602,8 @@ def test_jumpstart_model_package_arn(
             },
         )
 
+        self.assertIn(tag, mock_session.return_value.create_model.call_args[1]["tags"])
+
     @mock.patch("sagemaker.jumpstart.model.is_valid_model_id")
     @mock.patch("sagemaker.jumpstart.factory.model.Session")
     @mock.patch("sagemaker.jumpstart.accessors.JumpStartModelsAccessor.get_model_specs")

tests/unit/sagemaker/model/test_deploy.py

@@ -159,7 +159,12 @@ def test_deploy_accelerator_type(
         accelerator_type=ACCELERATOR_TYPE,
     )
 
-    create_sagemaker_model.assert_called_with(INSTANCE_TYPE, ACCELERATOR_TYPE, None, None)
+    create_sagemaker_model.assert_called_with(
+        instance_type=INSTANCE_TYPE,
+        accelerator_type=ACCELERATOR_TYPE,
+        tags=None,
+        serverless_inference_config=None,
+    )
     production_variant.assert_called_with(
         MODEL_NAME,
         INSTANCE_TYPE,
@@ -271,7 +276,12 @@ def test_deploy_tags(create_sagemaker_model, production_variant, name_from_base,
     tags = [{"Key": "ModelName", "Value": "TestModel"}]
     model.deploy(instance_type=INSTANCE_TYPE, initial_instance_count=INSTANCE_COUNT, tags=tags)
 
-    create_sagemaker_model.assert_called_with(INSTANCE_TYPE, None, tags, None)
+    create_sagemaker_model.assert_called_with(
+        instance_type=INSTANCE_TYPE,
+        accelerator_type=None,
+        tags=tags,
+        serverless_inference_config=None,
+    )
     sagemaker_session.endpoint_from_production_variants.assert_called_with(
         name=ENDPOINT_NAME,
         production_variants=[BASE_PRODUCTION_VARIANT],
@@ -463,7 +473,12 @@ def test_deploy_serverless_inference(production_variant, create_sagemaker_model,
         serverless_inference_config=serverless_inference_config,
     )
 
-    create_sagemaker_model.assert_called_with(None, None, None, serverless_inference_config)
+    create_sagemaker_model.assert_called_with(
+        instance_type=None,
+        accelerator_type=None,
+        tags=None,
+        serverless_inference_config=serverless_inference_config,
+    )
     production_variant.assert_called_with(
         MODEL_NAME,
         None,

tests/unit/sagemaker/model/test_model_package.py

@@ -197,7 +197,7 @@ def test_create_sagemaker_model_include_tags(sagemaker_session):
         sagemaker_session=sagemaker_session,
     )
 
-    model_package._create_sagemaker_model(tags=tags)
+    model_package.deploy(tags=tags, instance_type="ml.p2.xlarge", initial_instance_count=1)
 
     sagemaker_session.create_model.assert_called_with(
         model_name,

tests/unit/sagemaker/monitor/test_clarify_model_monitor.py

@@ -221,6 +221,14 @@
     "NetworkConfig": NETWORK_CONFIG._to_request_dict(),
 }
 
+MONITORING_EXECUTIONS_EMPTY = {
+    "MonitoringExecutionSummaries": [],
+}
+
+MONITORING_EXECUTIONS_NO_PROCESSING_JOB = {
+    "MonitoringExecutionSummaries": [{"MonitoringSchedule": "MonitoringSchedule"}],
+}
+
 # For update API
 NEW_ROLE_ARN = "arn:aws:iam::012345678902:role/{}".format(ROLE)
 NEW_INSTANCE_COUNT = 2
@@ -1716,3 +1724,20 @@ def _test_model_explainability_monitor_delete_schedule(
     sagemaker_session.sagemaker_client.delete_model_explainability_job_definition.assert_called_once_with(
         JobDefinitionName=job_definition_name
     )
+
+
+def test_model_explainability_monitor_logs_failure(model_explainability_monitor, sagemaker_session):
+    sagemaker_session.list_monitoring_executions = MagicMock(
+        return_value=MONITORING_EXECUTIONS_EMPTY
+    )
+    try:
+        model_explainability_monitor.get_latest_execution_logs()
+    except ValueError as ve:
+        assert "No execution jobs were kicked off." in str(ve)
+    sagemaker_session.list_monitoring_executions = MagicMock(
+        return_value=MONITORING_EXECUTIONS_NO_PROCESSING_JOB
+    )
+    try:
+        model_explainability_monitor.get_latest_execution_logs()
+    except ValueError as ve:
+        assert "Processing Job did not run for the last execution" in str(ve)