Merge branch 'master' into serverless-support-cr

Author: navinsoni

setup.py

@@ -16,7 +16,7 @@
 import os
 from glob import glob
 
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
 
 def read(fname):
@@ -81,6 +81,7 @@ def read_version():
         "fabric==2.6.0",
         "requests==2.27.1",
         "sagemaker-experiments==0.1.35",
+        "Jinja2==3.0.3",
     ],
 )
 

src/sagemaker/estimator.py

@@ -74,9 +74,7 @@
     get_config_value,
     name_from_base,
 )
-from sagemaker.workflow.entities import Expression
-from sagemaker.workflow.parameters import Parameter
-from sagemaker.workflow.properties import Properties
+from sagemaker.workflow.entities import PipelineVariable
 
 logger = logging.getLogger(__name__)
 
@@ -602,7 +600,7 @@ def _json_encode_hyperparameters(hyperparameters: Dict[str, Any]) -> Dict[str, A
         current_hyperparameters = hyperparameters
         if current_hyperparameters is not None:
             hyperparameters = {
-                str(k): (v if isinstance(v, (Parameter, Expression, Properties)) else json.dumps(v))
+                str(k): (v.to_string() if isinstance(v, PipelineVariable) else json.dumps(v))
                 for (k, v) in current_hyperparameters.items()
             }
         return hyperparameters
@@ -1813,7 +1811,7 @@ def _get_train_args(cls, estimator, inputs, experiment_config):
         current_hyperparameters = estimator.hyperparameters()
         if current_hyperparameters is not None:
             hyperparameters = {
-                str(k): (v if isinstance(v, (Parameter, Expression, Properties)) else str(v))
+                str(k): (v.to_string() if isinstance(v, PipelineVariable) else str(v))
                 for (k, v) in current_hyperparameters.items()
             }
 

src/sagemaker/model_monitor/clarify_model_monitoring.py

@@ -397,7 +397,6 @@ def _build_create_job_definition_request(
 
         if network_config is not None:
             network_config_dict = network_config._to_request_dict()
-            self._validate_network_config(network_config_dict)
             request_dict["NetworkConfig"] = network_config_dict
         elif existing_network_config is not None:
             request_dict["NetworkConfig"] = existing_network_config

src/sagemaker/model_monitor/model_monitoring.py

@@ -295,7 +295,6 @@ def create_monitoring_schedule(
         network_config_dict = None
         if self.network_config is not None:
             network_config_dict = self.network_config._to_request_dict()
-            self._validate_network_config(network_config_dict)
 
         self.sagemaker_session.create_monitoring_schedule(
             monitoring_schedule_name=self.monitoring_schedule_name,
@@ -448,7 +447,6 @@ def update_monitoring_schedule(
         network_config_dict = None
         if self.network_config is not None:
             network_config_dict = self.network_config._to_request_dict()
-            self._validate_network_config(network_config_dict)
 
         self.sagemaker_session.update_monitoring_schedule(
             monitoring_schedule_name=self.monitoring_schedule_name,
@@ -708,6 +706,9 @@ def attach(cls, monitor_schedule_name, sagemaker_session=None):
         if network_config_dict:
             network_config = NetworkConfig(
                 enable_network_isolation=network_config_dict["EnableNetworkIsolation"],
+                encrypt_inter_container_traffic=network_config_dict[
+                    "EnableInterContainerTrafficEncryption"
+                ],
                 security_group_ids=security_group_ids,
                 subnets=subnets,
             )
@@ -784,6 +785,9 @@ def _attach(clazz, sagemaker_session, schedule_desc, job_desc, tags):
         if network_config_dict:
             network_config = NetworkConfig(
                 enable_network_isolation=network_config_dict["EnableNetworkIsolation"],
+                encrypt_inter_container_traffic=network_config_dict[
+                    "EnableInterContainerTrafficEncryption"
+                ],
                 security_group_ids=security_group_ids,
                 subnets=subnets,
             )
@@ -1164,31 +1168,6 @@ def _wait_for_schedule_changes_to_apply(self):
             if schedule_desc["MonitoringScheduleStatus"] != "Pending":
                 break
 
-    def _validate_network_config(self, network_config_dict):
-        """Function to validate EnableInterContainerTrafficEncryption.
-
-        It validates EnableInterContainerTrafficEncryption is not set in the provided
-        NetworkConfig request dictionary.
-
-        Args:
-            network_config_dict (dict): NetworkConfig request dictionary.
-                Contains parameters from :class:`~sagemaker.network.NetworkConfig` object
-                that configures network isolation, encryption of
-                inter-container traffic, security group IDs, and subnets.
-
-        """
-        if "EnableInterContainerTrafficEncryption" in network_config_dict:
-            message = (
-                "EnableInterContainerTrafficEncryption is not supported in Model Monitor. "
-                "Please ensure that encrypt_inter_container_traffic=None "
-                "when creating your NetworkConfig object. "
-                "Current encrypt_inter_container_traffic value: {}".format(
-                    self.network_config.encrypt_inter_container_traffic
-                )
-            )
-            _LOGGER.info(message)
-            raise ValueError(message)
-
     @classmethod
     def monitoring_type(cls):
         """Type of the monitoring job."""
@@ -1781,7 +1760,6 @@ def update_monitoring_schedule(
         network_config_dict = None
         if self.network_config is not None:
             network_config_dict = self.network_config._to_request_dict()
-            super(DefaultModelMonitor, self)._validate_network_config(network_config_dict)
 
         if role is not None:
             self.role = role
@@ -2034,6 +2012,9 @@ def attach(cls, monitor_schedule_name, sagemaker_session=None):
             subnets = vpc_config.get("Subnets")
             network_config = NetworkConfig(
                 enable_network_isolation=network_config_dict["EnableNetworkIsolation"],
+                encrypt_inter_container_traffic=network_config_dict[
+                    "EnableInterContainerTrafficEncryption"
+                ],
                 security_group_ids=security_group_ids,
                 subnets=subnets,
             )
@@ -2304,7 +2285,6 @@ def _build_create_data_quality_job_definition_request(
 
         if network_config is not None:
             network_config_dict = network_config._to_request_dict()
-            self._validate_network_config(network_config_dict)
             request_dict["NetworkConfig"] = network_config_dict
         elif existing_network_config is not None:
             request_dict["NetworkConfig"] = existing_network_config
@@ -3007,7 +2987,6 @@ def _build_create_model_quality_job_definition_request(
 
         if network_config is not None:
             network_config_dict = network_config._to_request_dict()
-            self._validate_network_config(network_config_dict)
             request_dict["NetworkConfig"] = network_config_dict
         elif existing_network_config is not None:
             request_dict["NetworkConfig"] = existing_network_config

src/sagemaker/parameter.py

@@ -14,9 +14,8 @@
 from __future__ import absolute_import
 
 import json
-from sagemaker.workflow.parameters import Parameter as PipelineParameter
-from sagemaker.workflow.functions import JsonGet as PipelineJsonGet
-from sagemaker.workflow.functions import Join as PipelineJoin
+
+from sagemaker.workflow.entities import PipelineVariable
 
 
 class ParameterRange(object):
@@ -73,11 +72,11 @@ def as_tuning_range(self, name):
         return {
             "Name": name,
             "MinValue": str(self.min_value)
-            if not isinstance(self.min_value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else self.min_value,
+            if not isinstance(self.min_value, PipelineVariable)
+            else self.min_value.to_string(),
             "MaxValue": str(self.max_value)
-            if not isinstance(self.max_value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else self.max_value,
+            if not isinstance(self.max_value, PipelineVariable)
+            else self.max_value.to_string(),
             "ScalingType": self.scaling_type,
         }
 
@@ -112,8 +111,7 @@ def __init__(self, values):  # pylint: disable=super-init-not-called
         """
         values = values if isinstance(values, list) else [values]
         self.values = [
-            str(v) if not isinstance(v, (PipelineParameter, PipelineJsonGet, PipelineJoin)) else v
-            for v in values
+            str(v) if not isinstance(v, PipelineVariable) else v.to_string() for v in values
         ]
 
     def as_tuning_range(self, name):

src/sagemaker/tuner.py

@@ -38,6 +38,7 @@
     IntegerParameter,
     ParameterRange,
 )
+from sagemaker.workflow.entities import PipelineVariable
 from sagemaker.workflow.parameters import Parameter as PipelineParameter
 from sagemaker.workflow.functions import JsonGet as PipelineJsonGet
 from sagemaker.workflow.functions import Join as PipelineJoin
@@ -376,9 +377,7 @@ def _prepare_static_hyperparameters(
         """Prepare static hyperparameters for one estimator before tuning."""
         # Remove any hyperparameter that will be tuned
         static_hyperparameters = {
-            str(k): str(v)
-            if not isinstance(v, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else v
+            str(k): str(v) if not isinstance(v, PipelineVariable) else v.to_string()
             for (k, v) in estimator.hyperparameters().items()
         }
         for hyperparameter_name in hyperparameter_ranges.keys():

src/sagemaker/workflow/entities.py

@@ -16,7 +16,7 @@
 import abc
 
 from enum import EnumMeta
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Union, Optional
 
 PrimitiveType = Union[str, int, bool, float, None]
 RequestType = Union[Dict[str, Any], List[Dict[str, Any]]]
@@ -57,3 +57,80 @@ class Expression(abc.ABC):
     @abc.abstractmethod
     def expr(self) -> RequestType:
         """Get the expression structure for workflow service calls."""
+
+
+class PipelineVariable(Expression):
+    """Base object for pipeline variables
+
+    PipelineVariables must implement the expr property.
+    """
+
+    def __add__(self, other: Union[Expression, PrimitiveType]):
+        """Add function for PipelineVariable
+
+        Args:
+            other (Union[Expression, PrimitiveType]): The other object to be concatenated.
+
+        Always raise an error since pipeline variables do not support concatenation
+        """
+
+        raise TypeError("Pipeline variables do not support concatenation.")
+
+    def __str__(self):
+        """Override built-in String function for PipelineVariable"""
+        raise TypeError("Pipeline variables do not support __str__ operation.")
+
+    def __int__(self):
+        """Override built-in Integer function for PipelineVariable"""
+        raise TypeError("Pipeline variables do not support __int__ operation.")
+
+    def __float__(self):
+        """Override built-in Float function for PipelineVariable"""
+        raise TypeError("Pipeline variables do not support __float__ operation.")
+
+    def to_string(self):
+        """Prompt the pipeline to convert the pipeline variable to String in runtime"""
+        from sagemaker.workflow.functions import Join
+
+        return Join(on="", values=[self])
+
+    @property
+    @abc.abstractmethod
+    def expr(self) -> RequestType:
+        """Get the expression structure for workflow service calls."""
+
+    def startswith(
+        self,
+        prefix: Union[str, tuple],  # pylint: disable=unused-argument
+        start: Optional[int] = None,  # pylint: disable=unused-argument
+        end: Optional[int] = None,  # pylint: disable=unused-argument
+    ) -> bool:
+        """Simulate the Python string's built-in method: startswith
+
+        Args:
+            prefix (str, tuple): The (tuple of) string to be checked.
+            start (int): To set the start index of the matching boundary (default: None).
+            end (int): To set the end index of the matching boundary (default: None).
+
+        Return:
+            bool: Always return False as Pipeline variables are parsed during execution runtime
+        """
+        return False
+
+    def endswith(
+        self,
+        suffix: Union[str, tuple],  # pylint: disable=unused-argument
+        start: Optional[int] = None,  # pylint: disable=unused-argument
+        end: Optional[int] = None,  # pylint: disable=unused-argument
+    ) -> bool:
+        """Simulate the Python string's built-in method: endswith
+
+        Args:
+            suffix (str, tuple): The (tuple of) string to be checked.
+            start (int): To set the start index of the matching boundary (default: None).
+            end (int): To set the end index of the matching boundary (default: None).
+
+        Return:
+            bool: Always return False as Pipeline variables are parsed during execution runtime
+        """
+        return False

src/sagemaker/workflow/execution_variables.py

@@ -14,12 +14,12 @@
 from __future__ import absolute_import
 
 from sagemaker.workflow.entities import (
-    Expression,
     RequestType,
+    PipelineVariable,
 )
 
 
-class ExecutionVariable(Expression):
+class ExecutionVariable(PipelineVariable):
     """Pipeline execution variables for workflow."""
 
     def __init__(self, name: str):
@@ -30,6 +30,13 @@ def __init__(self, name: str):
         """
         self.name = name
 
+    def to_string(self) -> PipelineVariable:
+        """Prompt the pipeline to convert the pipeline variable to String in runtime
+
+        As ExecutionVariable is treated as String in runtime, no extra actions are needed.
+        """
+        return self
+
     @property
     def expr(self) -> RequestType:
         """The 'Get' expression dict for an `ExecutionVariable`."""

src/sagemaker/workflow/functions.py

@@ -17,12 +17,12 @@
 
 import attr
 
-from sagemaker.workflow.entities import Expression
+from sagemaker.workflow.entities import PipelineVariable
 from sagemaker.workflow.properties import PropertyFile
 
 
 @attr.s
-class Join(Expression):
+class Join(PipelineVariable):
     """Join together properties.
 
     Examples:
@@ -38,15 +38,23 @@ class Join(Expression):
     Attributes:
         values (List[Union[PrimitiveType, Parameter, Expression]]):
             The primitive type values, parameters, step properties, expressions to join.
-        on_str (str): The string to join the values on (Defaults to "").
+        on (str): The string to join the values on (Defaults to "").
     """
 
     on: str = attr.ib(factory=str)
     values: List = attr.ib(factory=list)
 
+    def to_string(self) -> PipelineVariable:
+        """Prompt the pipeline to convert the pipeline variable to String in runtime
+
+        As Join is treated as String in runtime, no extra actions are needed.
+        """
+        return self
+
     @property
     def expr(self):
         """The expression dict for a `Join` function."""
+
         return {
             "Std:Join": {
                 "On": self.on,
@@ -58,7 +66,7 @@ def expr(self):
 
 
 @attr.s
-class JsonGet(Expression):
+class JsonGet(PipelineVariable):
     """Get JSON properties from PropertyFiles.
 
     Attributes: