Merge branch 'master' into iss96247

Author: ahsan-z-khan

CHANGELOG.md

@@ -1,5 +1,35 @@
 # Changelog
 
+## v2.25.0 (2021-02-19)
+
+### Features
+
+ * Enable step caching
+ * Add other Neo supported regions for Inferentia inference images
+
+### Bug Fixes and Other Changes
+
+ * remove FailStep from pipelines
+ * use sagemaker_session in workflow tests
+ * use ECR public for multidatamodel tests
+ * add the mapping from py3 to cuda11 images
+ * Add 30s cap time for tag tests
+ * add build spec for slow tests
+ * mark top 10 slow tests
+ * remove slow test_run_xxx_monitor_baseline tests
+ * pin astroid to 2.4.2
+
+### Testing and Release Infrastructure
+
+ * unmark more flaky integ tests
+ * remove canary_quick pytest mark from flaky/unnecessary tests
+ * remove python3.8 from buildspec
+ * remove py38 tox env
+ * fix release buildspec typo
+ * unblock regional release builds
+ * lower test TPS for experiment analytics
+ * move package preparation and publishing to the deploy step
+
 ## v2.24.5 (2021-02-12)
 
 ### Bug Fixes and Other Changes

VERSION

@@ -1 +1 @@
-2.24.6.dev0
+2.25.1.dev0

src/sagemaker/workflow/steps.py

@@ -92,6 +92,38 @@ def ref(self) -> Dict[str, str]:
         return {"Name": self.name}
 
 
+@attr.s
+class CacheConfig:
+    """Configuration class to enable caching in pipeline workflow.
+
+    If caching is enabled, the pipeline attempts to find a previous execution of a step
+    that was called with the same arguments. Step caching only considers successful execution.
+    If a successful previous execution is found, the pipeline propagates the values
+    from previous execution rather than recomputing the step. When multiple successful executions
+    exist within the timeout period, it uses the result for the most recent successful execution.
+
+
+    Attributes:
+        enable_caching (bool): To enable step caching. Defaults to `False`.
+        expire_after (str): If step caching is enabled, a timeout also needs to defined.
+            It defines how old a previous execution can be to be considered for reuse.
+            Value should be an ISO 8601 duration string. Defaults to `None`.
+    """
+
+    enable_caching: bool = attr.ib(default=False)
+    expire_after = attr.ib(
+        default=None, validator=attr.validators.optional(attr.validators.instance_of(str))
+    )
+
+    @property
+    def config(self):
+        """Configures caching in pipeline steps."""
+        config = {"Enabled": self.enable_caching}
+        if self.expire_after is not None:
+            config["ExpireAfter"] = self.expire_after
+        return {"CacheConfig": config}
+
+
 class TrainingStep(Step):
     """Training step for workflow."""
 
@@ -100,6 +132,7 @@ def __init__(
         name: str,
         estimator: EstimatorBase,
         inputs: TrainingInput = None,
+        cache_config: CacheConfig = None,
     ):
         """Construct a TrainingStep, given an `EstimatorBase` instance.
 
@@ -110,14 +143,15 @@ def __init__(
             name (str): The name of the training step.
             estimator (EstimatorBase): A `sagemaker.estimator.EstimatorBase` instance.
             inputs (TrainingInput): A `sagemaker.inputs.TrainingInput` instance. Defaults to `None`.
+            cache_config (CacheConfig):  A `sagemaker.workflow.steps.CacheConfig` instance.
         """
         super(TrainingStep, self).__init__(name, StepTypeEnum.TRAINING)
         self.estimator = estimator
         self.inputs = inputs
-
         self._properties = Properties(
             path=f"Steps.{name}", shape_name="DescribeTrainingJobResponse"
         )
+        self.cache_config = cache_config
 
     @property
     def arguments(self) -> RequestType:
@@ -144,6 +178,14 @@ def properties(self):
         """A Properties object representing the DescribeTrainingJobResponse data model."""
         return self._properties
 
+    def to_request(self) -> RequestType:
+        """Updates the dictionary with cache configuration."""
+        request_dict = super().to_request()
+        if self.cache_config:
+            request_dict.update(self.cache_config.config)
+
+        return request_dict
+
 
 class CreateModelStep(Step):
     """CreateModel step for workflow."""
@@ -207,6 +249,7 @@ def __init__(
         name: str,
         transformer: Transformer,
         inputs: TransformInput,
+        cache_config: CacheConfig = None,
     ):
         """Constructs a TransformStep, given an `Transformer` instance.
 
@@ -217,11 +260,12 @@ def __init__(
             name (str): The name of the transform step.
             transformer (Transformer): A `sagemaker.transformer.Transformer` instance.
             inputs (TransformInput): A `sagemaker.inputs.TransformInput` instance.
+            cache_config (CacheConfig):  A `sagemaker.workflow.steps.CacheConfig` instance.
         """
         super(TransformStep, self).__init__(name, StepTypeEnum.TRANSFORM)
         self.transformer = transformer
         self.inputs = inputs
-
+        self.cache_config = cache_config
         self._properties = Properties(
             path=f"Steps.{name}", shape_name="DescribeTransformJobResponse"
         )
@@ -257,6 +301,14 @@ def properties(self):
         """A Properties object representing the DescribeTransformJobResponse data model."""
         return self._properties
 
+    def to_request(self) -> RequestType:
+        """Updates the dictionary with cache configuration."""
+        request_dict = super().to_request()
+        if self.cache_config:
+            request_dict.update(self.cache_config.config)
+
+        return request_dict
+
 
 class ProcessingStep(Step):
     """Processing step for workflow."""
@@ -270,6 +322,7 @@ def __init__(
         job_arguments: List[str] = None,
         code: str = None,
         property_files: List[PropertyFile] = None,
+        cache_config: CacheConfig = None,
     ):
         """Construct a ProcessingStep, given a `Processor` instance.
 
@@ -289,6 +342,7 @@ def __init__(
                 script to run. Defaults to `None`.
             property_files (List[PropertyFile]): A list of property files that workflow looks
                 for and resolves from the configured processing output list.
+            cache_config (CacheConfig):  A `sagemaker.workflow.steps.CacheConfig` instance.
         """
         super(ProcessingStep, self).__init__(name, StepTypeEnum.PROCESSING)
         self.processor = processor
@@ -305,6 +359,7 @@ def __init__(
         self._properties = Properties(
             path=f"Steps.{name}", shape_name="DescribeProcessingJobResponse"
         )
+        self.cache_config = cache_config
 
     @property
     def arguments(self) -> RequestType:
@@ -335,6 +390,8 @@ def properties(self):
     def to_request(self) -> RequestType:
         """Get the request structure for workflow service calls."""
         request_dict = super(ProcessingStep, self).to_request()
+        if self.cache_config:
+            request_dict.update(self.cache_config.config)
         if self.property_files:
             request_dict["PropertyFiles"] = [
                 property_file.expr for property_file in self.property_files

tests/integ/test_airflow_config.py

@@ -60,7 +60,6 @@
 SINGLE_INSTANCE_COUNT = 1
 
 
-@pytest.mark.canary_quick
 def test_byo_airflow_config_uploads_data_source_to_s3_when_inputs_provided(
     sagemaker_session, cpu_instance_type
 ):
@@ -92,7 +91,6 @@ def test_byo_airflow_config_uploads_data_source_to_s3_when_inputs_provided(
         )
 
 
-@pytest.mark.canary_quick
 def test_kmeans_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_instance_type):
     with timeout(seconds=AIRFLOW_CONFIG_TIMEOUT_IN_SECONDS):
         kmeans = KMeans(
@@ -153,7 +151,6 @@ def test_fm_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_inst
         )
 
 
-@pytest.mark.canary_quick
 def test_ipinsights_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_instance_type):
     with timeout(seconds=AIRFLOW_CONFIG_TIMEOUT_IN_SECONDS):
         data_path = os.path.join(DATA_DIR, "ipinsights")
@@ -214,7 +211,6 @@ def test_knn_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_ins
     tests.integ.test_region() in tests.integ.NO_LDA_REGIONS,
     reason="LDA image is not supported in certain regions",
 )
-@pytest.mark.canary_quick
 def test_lda_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_instance_type):
     with timeout(seconds=AIRFLOW_CONFIG_TIMEOUT_IN_SECONDS):
         data_path = os.path.join(DATA_DIR, "lda")
@@ -247,7 +243,6 @@ def test_lda_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_ins
         )
 
 
-@pytest.mark.canary_quick
 def test_linearlearner_airflow_config_uploads_data_source_to_s3(
     sagemaker_session, cpu_instance_type
 ):
@@ -312,7 +307,6 @@ def test_linearlearner_airflow_config_uploads_data_source_to_s3(
         )
 
 
-@pytest.mark.canary_quick
 def test_ntm_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_instance_type):
     with timeout(seconds=AIRFLOW_CONFIG_TIMEOUT_IN_SECONDS):
         data_path = os.path.join(DATA_DIR, "ntm")
@@ -346,7 +340,6 @@ def test_ntm_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_ins
         )
 
 
-@pytest.mark.canary_quick
 def test_pca_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_instance_type):
     with timeout(seconds=AIRFLOW_CONFIG_TIMEOUT_IN_SECONDS):
         pca = PCA(
@@ -373,7 +366,6 @@ def test_pca_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_ins
         )
 
 
-@pytest.mark.canary_quick
 def test_rcf_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_instance_type):
     with timeout(seconds=AIRFLOW_CONFIG_TIMEOUT_IN_SECONDS):
         # Generate a thousand 14-dimensional datapoints.
@@ -402,7 +394,6 @@ def test_rcf_airflow_config_uploads_data_source_to_s3(sagemaker_session, cpu_ins
         )
 
 
-@pytest.mark.canary_quick
 def test_chainer_airflow_config_uploads_data_source_to_s3(
     sagemaker_local_session, cpu_instance_type, chainer_latest_version, chainer_latest_py_version
 ):
@@ -440,7 +431,6 @@ def test_chainer_airflow_config_uploads_data_source_to_s3(
         )
 
 
-@pytest.mark.canary_quick
 def test_mxnet_airflow_config_uploads_data_source_to_s3(
     sagemaker_session,
     cpu_instance_type,
@@ -476,7 +466,6 @@ def test_mxnet_airflow_config_uploads_data_source_to_s3(
         )
 
 
-@pytest.mark.canary_quick
 def test_sklearn_airflow_config_uploads_data_source_to_s3(
     sagemaker_session,
     cpu_instance_type,
@@ -516,7 +505,6 @@ def test_sklearn_airflow_config_uploads_data_source_to_s3(
         )
 
 
-@pytest.mark.canary_quick
 def test_tf_airflow_config_uploads_data_source_to_s3(
     sagemaker_session,
     cpu_instance_type,
@@ -550,7 +538,6 @@ def test_tf_airflow_config_uploads_data_source_to_s3(
         )
 
 
-@pytest.mark.canary_quick
 def test_xgboost_airflow_config_uploads_data_source_to_s3(
     sagemaker_session, cpu_instance_type, xgboost_latest_version
 ):
@@ -576,7 +563,6 @@ def test_xgboost_airflow_config_uploads_data_source_to_s3(
         )
 
 
-@pytest.mark.canary_quick
 def test_pytorch_airflow_config_uploads_data_source_to_s3_when_inputs_not_provided(
     sagemaker_session,
     cpu_instance_type,

tests/integ/test_chainer.py

@@ -70,7 +70,6 @@ def test_training_with_additional_hyperparameters(
     chainer.fit({"train": train_input, "test": test_input})
 
 
-@pytest.mark.canary_quick
 def test_attach_deploy(
     sagemaker_session, chainer_latest_version, chainer_latest_py_version, cpu_instance_type
 ):

tests/integ/test_debugger.py

@@ -528,7 +528,6 @@ def test_mxnet_with_tensorboard_output_config(
         _wait_and_assert_that_no_rule_jobs_errored(training_job=mx.latest_training_job)
 
 
-@pytest.mark.canary_quick
 def test_mxnet_with_all_rules_and_configs(
     sagemaker_session,
     mxnet_training_latest_version,

tests/integ/test_mxnet.py

@@ -305,7 +305,6 @@ def test_deploy_model_and_update_endpoint(
         assert new_config["ProductionVariants"][0]["InitialInstanceCount"] == 1
 
 
-@pytest.mark.canary_quick
 @pytest.mark.skipif(
     tests.integ.test_region() not in tests.integ.EI_SUPPORTED_REGIONS,
     reason="EI isn't supported in that specific region.",

tests/integ/test_tuner_multi_algo.py

@@ -97,7 +97,6 @@ def estimator_knn(sagemaker_session, cpu_instance_type):
     return estimator
 
 
-@pytest.mark.canary_quick
 def test_multi_estimator_tuning(
     sagemaker_session, estimator_fm, estimator_knn, data_set, cpu_instance_type
 ):

tests/integ/test_workflow.py

@@ -46,6 +46,7 @@
     CreateModelStep,
     ProcessingStep,
     TrainingStep,
+    CacheConfig,
 )
 from sagemaker.workflow.step_collections import RegisterModel
 from sagemaker.workflow.pipeline import Pipeline
@@ -274,6 +275,8 @@ def test_one_step_sklearn_processing_pipeline(
         ProcessingInput(dataset_definition=athena_dataset_definition),
     ]
 
+    cache_config = CacheConfig(enable_caching=True, expire_after="T30m")
+
     sklearn_processor = SKLearnProcessor(
         framework_version=sklearn_latest_version,
         role=role,
@@ -289,6 +292,7 @@ def test_one_step_sklearn_processing_pipeline(
         processor=sklearn_processor,
         inputs=inputs,
         code=script_path,
+        cache_config=cache_config,
     )
     pipeline = Pipeline(
         name=pipeline_name,
@@ -328,6 +332,11 @@ def test_one_step_sklearn_processing_pipeline(
         response = execution.describe()
         assert response["PipelineArn"] == create_arn
 
+        # Check CacheConfig
+        response = json.loads(pipeline.describe()["PipelineDefinition"])["Steps"][0]["CacheConfig"]
+        assert response["Enabled"] == cache_config.enable_caching
+        assert response["ExpireAfter"] == cache_config.expire_after
+
         try:
             execution.wait(delay=30, max_attempts=3)
         except WaiterError: