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

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.

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

sagemaker-core/src/sagemaker/core/model_monitor/clarify_model_monitoring.py

@@ -31,7 +31,6 @@
 from sagemaker.core import image_uris, s3
 from sagemaker.core.helper.session_helper import Session, expand_role
 from sagemaker.core.common_utils import (
-    name_from_base,
     format_tags,
     get_resource_name_from_arn,
     list_tags,
@@ -640,7 +639,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,
@@ -783,7 +782,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,
@@ -1082,7 +1081,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,
@@ -1230,7 +1229,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,

sagemaker-core/src/sagemaker/core/model_monitor/model_monitoring.py

@@ -1213,6 +1213,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,
@@ -2074,7 +2090,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,
@@ -2444,7 +2460,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,
@@ -3208,7 +3224,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,
@@ -3364,7 +3380,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,

sagemaker-core/tests/unit/test_model_monitoring.py

@@ -380,6 +380,55 @@ def test_generate_baselining_job_name_with_custom_name(self, mock_session, test_
             job_name = monitor._generate_baselining_job_name(job_name="custom-job")
             assert job_name == "custom-job"
 
+    def test_generate_job_definition_name_uses_base_job_name(self, mock_session, test_role):
+        """A user-supplied base_job_name is reflected in the job definition name (#4783)."""
+        with (
+            patch(
+                "sagemaker.core.model_monitor.model_monitoring.resolve_value_from_config",
+                side_effect=lambda x, *args, **kwargs: x,
+            ),
+            patch(
+                "sagemaker.core.model_monitor.model_monitoring.resolve_class_attribute_from_config",
+                return_value=None,
+            ),
+            patch(
+                "sagemaker.core.model_monitor.model_monitoring."
+                "DefaultModelMonitor._get_default_image_uri",
+                return_value="test-image",
+            ),
+        ):
+            monitor = DefaultModelMonitor(
+                role=test_role,
+                sagemaker_session=mock_session,
+                base_job_name="my-monitor",
+            )
+            name = monitor._generate_job_definition_name()
+            assert name.startswith("my-monitor-")
+            assert DefaultModelMonitor.JOB_DEFINITION_BASE_NAME not in name
+
+    def test_generate_job_definition_name_defaults_without_base_job_name(
+        self, mock_session, test_role
+    ):
+        """Without base_job_name, the job definition name falls back to the type default."""
+        with (
+            patch(
+                "sagemaker.core.model_monitor.model_monitoring.resolve_value_from_config",
+                side_effect=lambda x, *args, **kwargs: x,
+            ),
+            patch(
+                "sagemaker.core.model_monitor.model_monitoring.resolve_class_attribute_from_config",
+                return_value=None,
+            ),
+            patch(
+                "sagemaker.core.model_monitor.model_monitoring."
+                "DefaultModelMonitor._get_default_image_uri",
+                return_value="test-image",
+            ),
+        ):
+            monitor = DefaultModelMonitor(role=test_role, sagemaker_session=mock_session)
+            name = monitor._generate_job_definition_name()
+            assert name.startswith(DefaultModelMonitor.JOB_DEFINITION_BASE_NAME)
+
     def test_start_monitoring_schedule(self, mock_session, test_role):
         with (
             patch(