fix(model-monitor): honor base_job_name for job definition names (#6373)

ModelMonitor subclasses generated the monitoring job definition name from
the type's JOB_DEFINITION_BASE_NAME constant, ignoring a user-supplied
base_job_name -- even though the monitoring schedule name already honors it.
As a result base_job_name was silently dropped for the job definition (and
the monitoring resources derived from it).

Add a _generate_job_definition_name() helper on the ModelMonitor base
(mirroring _generate_monitoring_schedule_name) that uses base_job_name when
provided and falls back to JOB_DEFINITION_BASE_NAME otherwise, and use it in
the data-quality, model-quality, model-bias and model-explainability create
and update paths.

Also make the four *_update_failure tests deterministic: they asserted the
regenerated job definition name differs from the original, which relied on
two name_from_base calls landing in different milliseconds (flaky on master).
Patch name_from_base with a counter-based unique name in those tests.

Note: the recurring execution processing job names are assigned by the
SageMaker service and remain non-customizable.

Relates-to #4783

Co-authored-by: Roja Reddy Sareddy <rsareddy@amazon.com>

Author: rsareddy0329

src/sagemaker/model_monitor/clarify_model_monitoring.py

@@ -25,7 +25,7 @@
 from sagemaker.model_monitor import model_monitoring as mm
 from sagemaker import image_uris, s3
 from sagemaker.session import Session
-from sagemaker.utils import name_from_base, format_tags
+from sagemaker.utils import format_tags
 from sagemaker.clarify import SageMakerClarifyProcessor, ModelPredictedLabelConfig
 from sagemaker.lineage._utils import get_resource_name_from_arn
 
@@ -625,7 +625,7 @@ def create_monitoring_schedule(
         monitor_schedule_name = self._generate_monitoring_schedule_name(
             schedule_name=monitor_schedule_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_job_definition_request(
             monitoring_schedule_name=monitor_schedule_name,
             job_definition_name=new_job_definition_name,
@@ -768,7 +768,7 @@ def update_monitoring_schedule(
         job_desc = self.sagemaker_session.sagemaker_client.describe_model_bias_job_definition(
             JobDefinitionName=self.job_definition_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_job_definition_request(
             monitoring_schedule_name=self.monitoring_schedule_name,
             job_definition_name=new_job_definition_name,
@@ -1067,7 +1067,7 @@ def create_monitoring_schedule(
         monitor_schedule_name = self._generate_monitoring_schedule_name(
             schedule_name=monitor_schedule_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_job_definition_request(
             monitoring_schedule_name=monitor_schedule_name,
             job_definition_name=new_job_definition_name,
@@ -1215,7 +1215,7 @@ def update_monitoring_schedule(
                 JobDefinitionName=self.job_definition_name
             )
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_job_definition_request(
             monitoring_schedule_name=self.monitoring_schedule_name,
             job_definition_name=new_job_definition_name,

src/sagemaker/model_monitor/model_monitoring.py

@@ -1180,6 +1180,22 @@ def _generate_monitoring_schedule_name(self, schedule_name=None):
 
         return name_from_base(base=base_name)
 
+    def _generate_job_definition_name(self):
+        """Generate the monitoring job definition name.
+
+        Uses the base name given to the constructor when provided, so a
+        user-supplied ``base_job_name`` is reflected in the job definition
+        name (and the monitoring resources derived from it) rather than being
+        silently ignored (see #4783); otherwise falls back to the monitor
+        type's default base name (``JOB_DEFINITION_BASE_NAME``).
+
+        Returns:
+            str: The generated job definition name.
+
+        """
+        base_name = self.base_job_name or self.JOB_DEFINITION_BASE_NAME
+        return name_from_base(base=base_name)
+
     @staticmethod
     def _generate_env_map(
         env,
@@ -2023,7 +2039,7 @@ def create_monitoring_schedule(
         monitor_schedule_name = self._generate_monitoring_schedule_name(
             schedule_name=monitor_schedule_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_data_quality_job_definition_request(
             monitoring_schedule_name=monitor_schedule_name,
             job_definition_name=new_job_definition_name,
@@ -2393,7 +2409,7 @@ def _update_data_quality_monitoring_schedule(
         job_desc = self.sagemaker_session.sagemaker_client.describe_data_quality_job_definition(
             JobDefinitionName=self.job_definition_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_data_quality_job_definition_request(
             monitoring_schedule_name=self.monitoring_schedule_name,
             job_definition_name=new_job_definition_name,
@@ -3147,7 +3163,7 @@ def create_monitoring_schedule(
         monitor_schedule_name = self._generate_monitoring_schedule_name(
             schedule_name=monitor_schedule_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_model_quality_job_definition_request(
             monitoring_schedule_name=monitor_schedule_name,
             job_definition_name=new_job_definition_name,
@@ -3303,7 +3319,7 @@ def update_monitoring_schedule(
         job_desc = self.sagemaker_session.sagemaker_client.describe_model_quality_job_definition(
             JobDefinitionName=self.job_definition_name
         )
-        new_job_definition_name = name_from_base(self.JOB_DEFINITION_BASE_NAME)
+        new_job_definition_name = self._generate_job_definition_name()
         request_dict = self._build_create_model_quality_job_definition_request(
             monitoring_schedule_name=self.monitoring_schedule_name,
             job_definition_name=new_job_definition_name,

tests/unit/sagemaker/monitor/test_clarify_model_monitor.py

@@ -402,6 +402,21 @@
 ] = ANALYSIS_CONFIG_LABEL_HEADERS
 
 
+def _unique_name_from_base(base, *args, **kwargs):
+    """Deterministic, collision-free stand-in for name_from_base in tests.
+
+    The real name_from_base appends a millisecond timestamp, so two calls in the
+    same millisecond (create + update within one test) can return identical names.
+    This counter-based helper keeps each generated name unique and still prefixed
+    by the base, which is what these tests assert.
+    """
+    _unique_name_from_base.counter += 1
+    return "{}-{:08d}".format(base, _unique_name_from_base.counter)
+
+
+_unique_name_from_base.counter = 0
+
+
 @pytest.fixture()
 def sagemaker_client():
     return MagicMock()
@@ -905,7 +920,13 @@ def test_model_bias_monitor_creation_failure(model_bias_monitor, sagemaker_sessi
     sagemaker_session.sagemaker_client.delete_model_bias_job_definition.assert_called_once()
 
 
-def test_model_bias_monitor_update_failure(model_bias_monitor, sagemaker_session):
+@patch(
+    "sagemaker.model_monitor.model_monitoring.name_from_base",
+    side_effect=_unique_name_from_base,
+)
+def test_model_bias_monitor_update_failure(
+    mock_name_from_base, model_bias_monitor, sagemaker_session
+):
     model_bias_monitor.create_monitoring_schedule(
         endpoint_input=ENDPOINT_NAME,
         ground_truth_input=GROUND_TRUTH_S3_URI,
@@ -1493,8 +1514,12 @@ def test_model_explainability_monitor_creation_failure(
     sagemaker_session.sagemaker_client.delete_model_explainability_job_definition.assert_called_once()
 
 
+@patch(
+    "sagemaker.model_monitor.model_monitoring.name_from_base",
+    side_effect=_unique_name_from_base,
+)
 def test_model_explainability_monitor_update_failure(
-    model_explainability_monitor, sagemaker_session
+    mock_name_from_base, model_explainability_monitor, sagemaker_session
 ):
     model_explainability_monitor.create_monitoring_schedule(
         endpoint_input=ENDPOINT_NAME,

tests/unit/sagemaker/monitor/test_model_monitoring.py

@@ -550,6 +550,25 @@ def data_quality_monitor(sagemaker_session):
     )
 
 
+def test_generate_job_definition_name_uses_base_job_name(data_quality_monitor):
+    """A user-supplied base_job_name is reflected in the job definition name (#4783)."""
+    name = data_quality_monitor._generate_job_definition_name()
+    assert name.startswith(BASE_JOB_NAME + "-")
+    assert DefaultModelMonitor.JOB_DEFINITION_BASE_NAME not in name
+
+
+def test_generate_job_definition_name_defaults_without_base_job_name(sagemaker_session):
+    """Without base_job_name, the job definition name falls back to the type default."""
+    monitor = DefaultModelMonitor(
+        role=ROLE,
+        instance_count=INSTANCE_COUNT,
+        instance_type=INSTANCE_TYPE,
+        sagemaker_session=sagemaker_session,
+    )
+    name = monitor._generate_job_definition_name()
+    assert name.startswith(DefaultModelMonitor.JOB_DEFINITION_BASE_NAME)
+
+
 @pytest.fixture()
 def model_monitor_arguments(sagemaker_session):
     return ModelMonitor(
@@ -953,7 +972,28 @@ def test_data_quality_monitor_invalid_attach(data_quality_monitor, sagemaker_ses
         )
 
 
-def test_data_quality_monitor_update_failure(data_quality_monitor, sagemaker_session):
+def _unique_name_from_base(base, *args, **kwargs):
+    """Deterministic, collision-free stand-in for name_from_base in tests.
+
+    The real name_from_base appends a millisecond timestamp, so two calls in the
+    same millisecond (create + update within one test) can return identical names.
+    This counter-based helper keeps each generated name unique and still prefixed
+    by the base, which is what these tests assert.
+    """
+    _unique_name_from_base.counter += 1
+    return "{}-{:08d}".format(base, _unique_name_from_base.counter)
+
+
+_unique_name_from_base.counter = 0
+
+
+@patch(
+    "sagemaker.model_monitor.model_monitoring.name_from_base",
+    side_effect=_unique_name_from_base,
+)
+def test_data_quality_monitor_update_failure(
+    mock_name_from_base, data_quality_monitor, sagemaker_session
+):
     data_quality_monitor.create_monitoring_schedule(
         endpoint_input=ENDPOINT_NAME,
     )
@@ -1623,7 +1663,13 @@ def test_model_quality_monitor_invalid_attach(model_quality_monitor, sagemaker_s
         )
 
 
-def test_model_quality_monitor_update_failure(model_quality_monitor, sagemaker_session):
+@patch(
+    "sagemaker.model_monitor.model_monitoring.name_from_base",
+    side_effect=_unique_name_from_base,
+)
+def test_model_quality_monitor_update_failure(
+    mock_name_from_base, model_quality_monitor, sagemaker_session
+):
     model_quality_monitor.create_monitoring_schedule(
         endpoint_input=ENDPOINT_NAME,
         ground_truth_input=GROUND_TRUTH_S3_URI,