feature: add a few minor features to Model Monitoring (aws#268)

* predictor.list_monitors() now returns a list of monitors.
* monitor.update/start/stop now waits for schedule to finish updating
before returning.
* monitor.create will fail if a schedule was already created,
explaining to the user that they must first delete a schedule before
creating a new one.
* fixed shallow copy on env to avoid mutating customer environment
dictionary
* fix integration tests based on latest API changes
* no longer relying on schedule defaults in ITs until API is updated.
* add missed imports to init to simplify user experience
* remove broken assert because API now randomizes schedule if not provided
* fix bug when attaching to monitor with entrypoint not specified

Author: knakad

src/sagemaker/estimator.py

@@ -615,8 +615,8 @@ def deploy(
             kms_key (str): The ARN of the KMS key that is used to encrypt the
                 data on the storage volume attached to the instance hosting the
                 endpoint.
-            data_capture_config (DataCaptureConfig): Specifies configuration
-                related to Endpoint data capture for use with
+            data_capture_config (sagemaker.model_monitor.DataCaptureConfig): Specifies
+                configuration related to Endpoint data capture for use with
                 Amazon SageMaker Model Monitoring. Default: None.
             **kwargs: Passed to invocation of ``create_model()``.
                 Implementations may customize ``create_model()`` to accept

src/sagemaker/model.py

@@ -425,8 +425,8 @@ def deploy(
                 endpoint.
             wait (bool): Whether the call should wait until the deployment of
                 this model completes (default: True).
-            data_capture_config (DataCaptureConfig): Specifies configuration
-                related to Endpoint data capture for use with
+            data_capture_config (sagemaker.model_monitor.DataCaptureConfig): Specifies
+                configuration related to Endpoint data capture for use with
                 Amazon SageMaker Model Monitoring. Default: None.
 
         Returns:
@@ -458,9 +458,9 @@ def deploy(
             if self._is_compiled_model and not self.endpoint_name.endswith(compiled_model_suffix):
                 self.endpoint_name += compiled_model_suffix
 
-        data_capture_config_dict = (
-            data_capture_config.to_request_dict() if data_capture_config else None
-        )
+        data_capture_config_dict = None
+        if data_capture_config is not None:
+            data_capture_config_dict = data_capture_config.to_request_dict()
 
         if update_endpoint:
             endpoint_config_name = self.sagemaker_session.create_endpoint_config(

src/sagemaker/model_monitor/__init__.py

@@ -20,6 +20,9 @@
 
 from sagemaker.model_monitor.model_monitoring import ModelMonitor  # noqa: F401
 from sagemaker.model_monitor.model_monitoring import DefaultModelMonitor  # noqa: F401
+from sagemaker.model_monitor.model_monitoring import BaseliningJob  # noqa: F401
+from sagemaker.model_monitor.model_monitoring import MonitoringExecution  # noqa: F401
+from sagemaker.model_monitor.model_monitoring import EndpointInput  # noqa: F401
 from sagemaker.model_monitor.model_monitoring import MonitoringOutput  # noqa: F401
 
 from sagemaker.model_monitor.cron_expression_generator import CronExpressionGenerator  # noqa: F401
@@ -28,5 +31,6 @@
 from sagemaker.model_monitor.monitoring_files import ConstraintViolations  # noqa: F401
 
 from sagemaker.model_monitor.data_capture_config import DataCaptureConfig  # noqa: F401
+from sagemaker.model_monitor.dataset_format import DatasetFormat  # noqa: F401
 
 from sagemaker.network import NetworkConfig  # noqa: F401

src/sagemaker/model_monitor/data_capture_config.py

@@ -49,7 +49,7 @@ def __init__(
             sampling_percentage (int): Optional. Default=20. The percentage of data to sample.
                 Must be between 0 and 100.
             destination_s3_uri (str): Optional. Defaults to "s3://<default-session-bucket>/
-                <model-monitor>/data-capture
+                model-monitor/data-capture".
             kms_key_id (str): Optional. Default=None. The kms key to use when writing to S3.
             capture_options ([str]): Optional. Must be a list containing any combination of the
                 following values: "REQUEST", "RESPONSE". Default=["REQUEST", "RESPONSE"]. Denotes
@@ -78,8 +78,9 @@ def to_request_dict(self):
             "InitialSamplingPercentage": self.sampling_percentage,
             "DestinationS3Uri": self.destination_s3_uri,
             "CaptureOptions": [
-                {"CaptureMode": dict(self.API_MAPPING).get(capture_option.upper(), capture_option)}
-                for capture_option in list(self.capture_options)
+                #  Convert to API values or pass value directly through if unable to convert.
+                {"CaptureMode": self.API_MAPPING.get(capture_option.upper(), capture_option)}
+                for capture_option in self.capture_options
             ],
         }
 

src/sagemaker/model_monitor/dataset_format.py

@@ -42,7 +42,7 @@ def json(lines=True):
         """Returns a DatasetFormat JSON string for use with a DefaultModelMonitor.
 
         Args:
-            lines (bool): Read the file as a json object per line. Default: True.
+            lines (bool): Whether the file should be read as a json object per line. Default: True.
 
         Returns:
             dict: JSON string containing DatasetFormat to be used by DefaultModelMonitor.

src/sagemaker/model_monitor/model_monitoring.py

@@ -10,12 +10,13 @@
 # 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.
-"""This module contains code related to Amazon SageMaker Monitoring Schedules. These
-classes assist with suggesting baselines and creating monitoring schedules for data captured
+"""This module contains code related to Amazon SageMaker Model Monitoring. These classes
+assist with suggesting baselines and creating monitoring schedules for data captured
 by SageMaker Endpoints.
 """
 from __future__ import print_function, absolute_import
 
+import copy
 import json
 import os
 import logging
@@ -35,10 +36,10 @@
 from sagemaker.processing import ProcessingJob
 from sagemaker.processing import ProcessingInput
 from sagemaker.processing import ProcessingOutput
-from sagemaker.model_monitor.cron_expression_generator import CronExpressionGenerator
 from sagemaker.model_monitor.monitoring_files import Constraints, ConstraintViolations
 from sagemaker.model_monitor.monitoring_files import Statistics
 from sagemaker.exceptions import UnexpectedStatusException
+from sagemaker.utils import retries
 
 _DEFAULT_MONITOR_IMAGE_URI_WITH_PLACEHOLDERS = (
     "{}.dkr.ecr.{}.amazonaws.com/sagemaker-model-monitor-analyzer"
@@ -58,8 +59,7 @@
     "ap-northeast-2": "709848358524",
     "eu-west-2": "749857270468",
     "ap-northeast-1": "574779866223",
-    "us-west-2": "159807026194",  # Prod
-    # "us-west-2": "894667893881",  # Gamma. # TODO-reinvent-2019 [knakad]: Remove this.
+    "us-west-2": "159807026194",
     "us-west-1": "890145073186",
     "ap-southeast-1": "245545462676",
     "ap-southeast-2": "563025443158",
@@ -86,7 +86,6 @@
 
 _SUGGESTION_JOB_BASE_NAME = "baseline-suggestion-job"
 _MONITORING_SCHEDULE_BASE_NAME = "monitoring-schedule"
-_SCHEDULE_NAME_SUFFIX = "monitoring-schedule"
 
 _DATASET_SOURCE_PATH_ENV_NAME = "dataset_source"
 _DATASET_FORMAT_ENV_NAME = "dataset_format"
@@ -96,7 +95,6 @@
 _PUBLISH_CLOUDWATCH_METRICS_ENV_NAME = "publish_cloudwatch_metrics"
 
 _LOGGER = logging.getLogger(__name__)
-# TODO-reinvent-2019 [knakad]: Review all docstrings.
 
 
 class ModelMonitor(object):
@@ -239,8 +237,8 @@ def create_monitoring_schedule(
         output,
         statistics=None,
         constraints=None,
-        monitor_schedule_name=None,  # TODO-reinvent-2019 [knakad]: Change to mon_sched_name evwhere
-        schedule_cron_expression=CronExpressionGenerator.hourly(),
+        monitor_schedule_name=None,
+        schedule_cron_expression=None,
     ):
         """Creates a monitoring schedule to monitor an Amazon SageMaker Endpoint.
 
@@ -266,9 +264,18 @@ def create_monitoring_schedule(
                 a default job name, based on the image name and current timestamp.
             schedule_cron_expression (str): The cron expression that dictates the frequency that
                 this job runs at. See sagemaker.model_monitor.CronExpressionGenerator for valid
-                expressions.
+                expressions. Default: Daily.
 
         """
+        if self.monitoring_schedule_name is not None:
+            message = (
+                "It seems that this object was already used to create an Amazon Model "
+                "Monitoring Schedule. To create another, first delete the existing one "
+                "using my_monitor.delete_monitoring_schedule()."
+            )
+            print(message)
+            raise ValueError(message)
+
         self.monitoring_schedule_name = self._generate_monitoring_schedule_name(
             schedule_name=monitor_schedule_name
         )
@@ -474,23 +481,30 @@ def update_monitoring_schedule(
             role_arn=role,
         )
 
+        self._wait_for_schedule_changes_to_apply()
+
     def start_monitoring_schedule(self):
         """Starts the monitoring schedule."""
         self.sagemaker_session.start_monitoring_schedule(
             monitoring_schedule_name=self.monitoring_schedule_name
         )
 
+        self._wait_for_schedule_changes_to_apply()
+
     def stop_monitoring_schedule(self):
         """Stops the monitoring schedule."""
         self.sagemaker_session.stop_monitoring_schedule(
             monitoring_schedule_name=self.monitoring_schedule_name
         )
 
+        self._wait_for_schedule_changes_to_apply()
+
     def delete_monitoring_schedule(self):
         """Deletes the monitoring schedule."""
         self.sagemaker_session.delete_monitoring_schedule(
             monitoring_schedule_name=self.monitoring_schedule_name
         )
+        self.monitoring_schedule_name = None
 
     def baseline_statistics(self, file_name=STATISTICS_JSON_DEFAULT_FILE_NAME):
         """Returns a Statistics object representing the statistics json file generated by the
@@ -665,7 +679,7 @@ def attach(cls, monitor_schedule_name, sagemaker_session=None):
         ]["ClusterConfig"]["InstanceType"]
         entrypoint = schedule_desc["MonitoringScheduleConfig"]["MonitoringJobDefinition"][
             "MonitoringAppSpecification"
-        ]["ContainerEntrypoint"]
+        ].get("ContainerEntrypoint")
         volume_size_in_gb = schedule_desc["MonitoringScheduleConfig"]["MonitoringJobDefinition"][
             "MonitoringResources"
         ]["ClusterConfig"]["VolumeSizeInGB"]
@@ -744,7 +758,7 @@ def _generate_baselining_job_name(self, job_name=None):
             return job_name
 
         if self.base_job_name:
-            base_name = "{}-{}".format(self.base_job_name, _SCHEDULE_NAME_SUFFIX)
+            base_name = self.base_job_name
         else:
             base_name = _SUGGESTION_JOB_BASE_NAME
 
@@ -932,6 +946,20 @@ def _s3_uri_from_local_path(self, path):
             path = os.path.join(s3_uri, os.path.basename(path))
         return path
 
+    def _wait_for_schedule_changes_to_apply(self):
+        """Waits for the schedule associated with this monitor to no longer be in the 'Pending'
+        state.
+
+        """
+        for _ in retries(
+            max_retry_count=36,  # 36*5 = 3min
+            exception_message_prefix="Waiting for schedule to leave 'Pending' status",
+            seconds_to_sleep=5,
+        ):
+            schedule_desc = self.describe_schedule()
+            if schedule_desc["MonitoringScheduleStatus"] != "Pending":
+                break
+
 
 class DefaultModelMonitor(ModelMonitor):
     """Sets up Amazon SageMaker Monitoring Schedules and baseline suggestions. Use this class when
@@ -1088,7 +1116,6 @@ def suggest_baseline(
             dataset_format=dataset_format,
             output_path=normalized_baseline_output.source,
             enable_cloudwatch_metrics=False,  # Only supported for monitoring schedules
-            # TODO-reinvent-2019 [knakad]: Remove this once API stops failing if not provided.
             dataset_source_container_path=baseline_dataset_container_path,
             record_preprocessor_script_container_path=record_preprocessor_script_container_path,
             post_processor_script_container_path=post_processor_script_container_path,
@@ -1147,8 +1174,7 @@ def create_monitoring_schedule(
         constraints=None,
         statistics=None,
         monitor_schedule_name=None,
-        schedule_cron_expression=CronExpressionGenerator.hourly(),
-        # TODO-reinvent-2019 [knakad]: Service to default this to daily at a random hour
+        schedule_cron_expression=None,
         enable_cloudwatch_metrics=True,
     ):
         """Creates a monitoring schedule to monitor an Amazon SageMaker Endpoint.
@@ -1179,11 +1205,20 @@ def create_monitoring_schedule(
                 a default job name, based on the image name and current timestamp.
             schedule_cron_expression (str): The cron expression that dictates the frequency that
                 this job run. See sagemaker.model_monitor.CronExpressionGenerator for valid
-                expressions.
+                expressions. Default: Daily.
             enable_cloudwatch_metrics (bool): Whether to publish cloudwatch metrics as part of
                 the baselining or monitoring jobs.
 
         """
+        if self.monitoring_schedule_name is not None:
+            message = (
+                "It seems that this object was already used to create an Amazon Model "
+                "Monitoring Schedule. To create another, first delete the existing one "
+                "using my_monitor.delete_monitoring_schedule()."
+            )
+            print(message)
+            raise ValueError(message)
+
         self.monitoring_schedule_name = self._generate_monitoring_schedule_name(
             schedule_name=monitor_schedule_name
         )
@@ -1354,12 +1389,7 @@ def update_monitoring_schedule(
             self.env = env
 
         normalized_env = self._generate_env_map(
-            env=env,
-            # dataset_format=DatasetFormat.sagemaker_capture_json(),
-            output_path=output_path,
-            enable_cloudwatch_metrics=enable_cloudwatch_metrics,
-            # record_preprocessor_script_input=record_preprocessor_script_input,
-            # post_analytics_processor_script_input=post_analytics_processor_script_input,
+            env=env, output_path=output_path, enable_cloudwatch_metrics=enable_cloudwatch_metrics
         )
 
         statistics_object, constraints_object = self._get_baseline_files(
@@ -1422,6 +1452,8 @@ def update_monitoring_schedule(
             role_arn=role,
         )
 
+        self._wait_for_schedule_changes_to_apply()
+
     def run_baseline(self):
         """'.run_baseline()' is only allowed for ModelMonitor objects. Please use suggest_baseline
         for DefaultModelMonitor objects, instead."""
@@ -1569,8 +1601,8 @@ def latest_monitoring_constraint_violations(self):
         except ClientError:
             status = latest_monitoring_execution.describe()["ProcessingJobStatus"]
             print(
-                "Unable to retrieve statistics as job is in status '{}'. Latest violations only "
-                "available for completed executions.".format(status)
+                "Unable to retrieve constraint violations as job is in status '{}'. Latest "
+                "violations only available for completed executions.".format(status)
             )
 
     def _normalize_baseline_output(self, output_s3_uri=None):
@@ -1649,7 +1681,7 @@ def _generate_env_map(
         cloudwatch_env_map = {True: "Enabled", False: "Disabled"}
 
         if env is not None:
-            env = env.copy()
+            env = copy.deepcopy(env)
         env = env or {}
 
         if output_path is not None:
@@ -1672,12 +1704,6 @@ def _generate_env_map(
         if dataset_source_container_path is not None:
             env[_DATASET_SOURCE_PATH_ENV_NAME] = dataset_source_container_path
 
-        # if dataset_source_input is not None:
-        #     dataset_source_input_container_path = os.path.join(
-        #         dataset_source_input.destination, os.path.basename(dataset_source_input.source)
-        #     )
-        #     env[_DATASET_SOURCE_PATH_ENV_NAME] = dataset_source_input_container_path
-
         return env
 
     def _upload_and_convert_to_processing_input(self, source, destination, name):
@@ -1808,7 +1834,7 @@ def baseline_statistics(self, file_name=STATISTICS_JSON_DEFAULT_FILE_NAME, kms_k
                         actual_status=status,
                     )
             else:
-                raise
+                raise client_error
 
     def suggested_constraints(self, file_name=CONSTRAINTS_JSON_DEFAULT_FILE_NAME, kms_key=None):
         """Returns a sagemaker.model_monitor.Constraints object representing the constraints
@@ -1845,7 +1871,7 @@ def suggested_constraints(self, file_name=CONSTRAINTS_JSON_DEFAULT_FILE_NAME, km
                         actual_status=status,
                     )
             else:
-                raise
+                raise client_error
 
 
 class MonitoringExecution(ProcessingJob):
@@ -1956,7 +1982,7 @@ def statistics(self, file_name=STATISTICS_JSON_DEFAULT_FILE_NAME, kms_key=None):
                         actual_status=status,
                     )
             else:
-                raise
+                raise client_error
 
     def constraint_violations(
         self, file_name=CONSTRAINT_VIOLATIONS_JSON_DEFAULT_FILE_NAME, kms_key=None
@@ -1997,7 +2023,7 @@ def constraint_violations(
                         actual_status=status,
                     )
             else:
-                raise
+                raise client_error
 
 
 class EndpointInput(object):