in source/lib/blueprints/byom/pipeline_definitions/iam_policies.py [0:0]
def sagemaker_monitor_policy_statement(baseline_job_name, monitoring_schedule_name, endpoint_name, monitoring_type):
# common permissions
actions = [
"sagemaker:DescribeEndpointConfig",
"sagemaker:DescribeEndpoint",
"sagemaker:CreateMonitoringSchedule",
"sagemaker:DescribeMonitoringSchedule",
"sagemaker:StopMonitoringSchedule",
"sagemaker:DeleteMonitoringSchedule",
"sagemaker:DescribeProcessingJob",
]
# common resources
resources = [
f"{sagemaker_arn_prefix}:endpoint-config/mlopssagemakerendpointconfig*",
f"{sagemaker_arn_prefix}:endpoint/{endpoint_name}",
f"{sagemaker_arn_prefix}:monitoring-schedule/{monitoring_schedule_name}",
f"{sagemaker_arn_prefix}:processing-job/{baseline_job_name}",
]
# create a map of monitoring type -> required permissions/resources
type_permissions = {
"DataQuality": {
"permissions": [
"sagemaker:CreateDataQualityJobDefinition",
"sagemaker:DescribeDataQualityJobDefinition",
"sagemaker:DeleteDataQualityJobDefinition",
],
"resources": [f"{sagemaker_arn_prefix}:data-quality-job-definition/*"],
},
"ModelQuality": {
"permissions": [
"sagemaker:CreateModelQualityJobDefinition",
"sagemaker:DescribeModelQualityJobDefinition",
"sagemaker:DeleteModelQualityJobDefinition",
],
"resources": [f"{sagemaker_arn_prefix}:model-quality-job-definition/*"],
},
"ModelBias": {
"permissions": [
"sagemaker:CreateModelBiasJobDefinition",
"sagemaker:DescribeModelBiasJobDefinition",
"sagemaker:DeleteModelBiasJobDefinition",
],
"resources": [f"{sagemaker_arn_prefix}:model-bias-job-definition/*"],
},
"ModelExplainability": {
"permissions": [
"sagemaker:CreateModelExplainabilityJobDefinition",
"sagemaker:DescribeModelExplainabilityJobDefinition",
"sagemaker:DeleteModelExplainabilityJobDefinition",
],
"resources": [f"{sagemaker_arn_prefix}:model-explainability-job-definition/*"],
},
}
# add monitoring type's specific permissions
actions.extend(type_permissions[monitoring_type]["permissions"])
# add monitoring type's specific resources
resources.extend(type_permissions[monitoring_type]["resources"])
# create the policy statement
return iam.PolicyStatement(
actions=actions,
resources=resources,
)