diff --git a/sdk/ml/azure-ai-ml/CHANGELOG.md b/sdk/ml/azure-ai-ml/CHANGELOG.md index bc3df33c6c49..0193086da0ad 100644 --- a/sdk/ml/azure-ai-ml/CHANGELOG.md +++ b/sdk/ml/azure-ai-ml/CHANGELOG.md @@ -5,6 +5,7 @@ ### Features Added ### Bugs Fixed +- Fixed `MLClient.jobs.create_or_update`, `archive`, and `restore` failing for previously-fetched jobs across all job types by routing metadata-only edits through the RunHistory PATCH endpoint. ### Other Changes diff --git a/sdk/ml/azure-ai-ml/api.md b/sdk/ml/azure-ai-ml/api.md index d339f6b6a5bc..576a38619750 100644 --- a/sdk/ml/azure-ai-ml/api.md +++ b/sdk/ml/azure-ai-ml/api.md @@ -10934,6 +10934,18 @@ namespace azure.ai.ml.operations @monitor_with_activity(ops_logger, 'Job.Stream', ActivityType.PUBLICAPI) def stream(self, name: str) -> None: ... + @distributed_trace + @monitor_with_telemetry_mixin(ops_logger, 'Job.Update', ActivityType.PUBLICAPI) + def update( + self, + name: str, + *, + description: Optional[str] = ..., + display_name: Optional[str] = ..., + properties: Optional[Dict[str, str]] = ..., + tags: Optional[Dict[str, str]] = ... + ) -> Job: ... + @distributed_trace @monitor_with_telemetry_mixin(ops_logger, 'Job.Validate', ActivityType.PUBLICAPI) def validate( diff --git a/sdk/ml/azure-ai-ml/api.metadata.yml b/sdk/ml/azure-ai-ml/api.metadata.yml index 3c170cd25905..9549c7c79293 100644 --- a/sdk/ml/azure-ai-ml/api.metadata.yml +++ b/sdk/ml/azure-ai-ml/api.metadata.yml @@ -1,3 +1,3 @@ -apiMdSha256: dc079c94f44c28406c444f39bd5501c997e56e593fd93c3176b08ca3c3ef04d3 +apiMdSha256: 7262033cadec8b494dfa9cca72d97a9dad84e1e465ce8c455782d9fab275a81f parserVersion: 0.3.28 pythonVersion: 3.12.10 diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py b/sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py index 386dafa2ca94..f5751ab1a1cc 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py +++ b/sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py @@ -654,7 +654,7 @@ def error_func(msg: str, no_personal_data_msg: str) -> ValidationException: @distributed_trace @monitor_with_telemetry_mixin(ops_logger, "Job.CreateOrUpdate", ActivityType.PUBLICAPI) - def create_or_update( + def create_or_update( # pylint: disable=too-many-branches self, job: Job, *, @@ -723,6 +723,67 @@ def create_or_update( if experiment_name is not None: job.experiment_name = experiment_name + # ---------------- TEMPORARY HOTFIX (metadata-only shortcut) ---------------- + # Root bug: for a Job fetched via jobs.get(name) then mutated and passed to + # create_or_update, the serialize + MFE PUT round-trip fails for AutoML, + # Command, Pipeline, Sweep, Spark (each with a different serializer / + # comparator error). Until MFE + per-jobType serializers are fixed upstream, + # detect the "fetched job with only metadata changes" case and route through + # the RunHistory PATCH endpoint (same call used by ML Studio's portal for + # inline edits, and by the new jobs.update() API added alongside this fix). + # + # Shortcut is taken only when: + # * the incoming job carries an ARM resource id (=> it was fetched, not + # freshly created), AND + # * the caller is not trying to change compute or experiment_name (these + # are not supported by the shortcut and were rejected by MFE anyway). + # Any failure inside the shortcut falls through to the original path so + # brand-new job creation behavior is preserved bit-for-bit. + if getattr(job, "id", None) and getattr(job, "name", None) and compute is None and experiment_name is None: + job_name: str = cast(str, job.name) + patch_succeeded = False + try: + job_object = self._get_job(job_name) + if job_object.properties.job_type == RestJobType.PIPELINE: + job_object = self._get_job_2401(job_name) + if _is_pipeline_child_job(job_object): + raise PipelineChildJobError(job_id=job_object.id) + + from azure.ai.ml._restclient.runhistory.models import CreateRun + + self._runs_operations._operation.add_or_modify_by_experiment_name( + subscription_id=self._operation_scope.subscription_id, + resource_group_name=self._operation_scope.resource_group_name, + workspace_name=self._workspace_name, + experiment_name=job_object.properties.experiment_name, + run_id=job_name, + body=CreateRun( + display_name=getattr(job, "display_name", None), + description=getattr(job, "description", None), + tags=getattr(job, "tags", None), + properties=getattr(job, "properties", None), + ), + ) + patch_succeeded = True + except PipelineChildJobError: + raise + except Exception: # pylint: disable=broad-exception-caught + # Only failures BEFORE and INCLUDING the RunHistory PATCH are swallowed + # (job not found, transient RunHistory error, etc.) so we can fall + # through to the original code path without regressing creation. + pass + + if patch_succeeded: + # PATCH already persisted the metadata change server-side. Any failure + # in the refresh/resolve below must surface directly — falling back to + # the original MFE PUT path would re-execute the very round-trip this + # hotfix exists to avoid and would surface a misleading error. + refreshed = self._get_job(job_name) + if refreshed.properties.job_type == RestJobType.PIPELINE: + refreshed = self._get_job_2401(job_name) + return self._resolve_azureml_id(Job._from_rest_object(refreshed)) + # ------------------ END TEMPORARY HOTFIX -------------------------------- + if job.compute == LOCAL_COMPUTE_TARGET: job.environment_variables[COMMON_RUNTIME_ENV_VAR] = "true" # type: ignore @@ -850,8 +911,34 @@ def _archive_or_restore(self, name: str, is_archived: bool) -> None: job_object = self._get_job_2401(name) if _is_pipeline_child_job(job_object): raise PipelineChildJobError(job_id=job_object.id) - job_object.properties.is_archived = is_archived + # ---------------- TEMPORARY HOTFIX (archive/restore shortcut) ---------------- + # The legacy _create_or_update_with_different_version_api PUT round-trip fails + # on AutoML, Command, Sweep, and Spark jobs for the same serializer/comparator + # reasons as create_or_update (see the shortcut in create_or_update above). + # RunHistory's "hidden" flag is the same server-side field surfaced by MFE as + # properties.isArchived (verified via probe_runhistory_hidden.py), so we can + # PATCH it directly - which is what ML Studio's portal does for the Archive + # button. Any failure falls through to the original path so behavior is + # preserved for Pipeline (which already works via _get_job_2401). + try: + from azure.ai.ml._restclient.runhistory.models import CreateRun + + self._runs_operations._operation.add_or_modify_by_experiment_name( + subscription_id=self._operation_scope.subscription_id, + resource_group_name=self._operation_scope.resource_group_name, + workspace_name=self._workspace_name, + experiment_name=job_object.properties.experiment_name, + run_id=name, + body=CreateRun(hidden=is_archived), + ) + return + except Exception: # pylint: disable=broad-exception-caught + # Fall through to the original code path on any RunHistory error. + pass + # ------------------ END TEMPORARY HOTFIX -------------------------------- + + job_object.properties.is_archived = is_archived self._create_or_update_with_different_version_api(rest_job_resource=job_object) @distributed_trace @@ -896,6 +983,71 @@ def restore(self, name: str) -> None: self._archive_or_restore(name=name, is_archived=False) + @distributed_trace + @monitor_with_telemetry_mixin(ops_logger, "Job.Update", ActivityType.PUBLICAPI) + def update( + self, + name: str, + *, + display_name: Optional[str] = None, + description: Optional[str] = None, + tags: Optional[Dict[str, str]] = None, + properties: Optional[Dict[str, str]] = None, + ) -> Job: + """Partially updates the display_name, description, tags or properties of an existing job. + + This routes through the RunHistory PATCH endpoint that Azure ML Studio's portal uses + for inline edits, which works uniformly for all job types (AutoML, Command, Pipeline, + Sweep, Spark, Parallel) and merges tags/properties rather than replacing them. + + :param name: The name of the job to update. + :type name: str + :keyword display_name: New display name for the job. If None, leaves the current value unchanged. + :paramtype display_name: Optional[str] + :keyword description: New description for the job. If None, leaves the current value unchanged. + :paramtype description: Optional[str] + :keyword tags: Tags to merge into the job's existing tags. + :paramtype tags: Optional[Dict[str, str]] + :keyword properties: Properties to merge into the job's existing properties. + :paramtype properties: Optional[Dict[str, str]] + :raises ~azure.ai.ml.exceptions.UserErrorException: Raised if no updatable field is supplied. + :raises ~azure.ai.ml.exceptions.PipelineChildJobError: Raised if the target job is a pipeline child job. + :return: The refreshed Job entity. + :rtype: ~azure.ai.ml.entities.Job + """ + if display_name is None and description is None and tags is None and properties is None: + raise UserErrorException( + "jobs.update() requires at least one of " + "display_name, description, tags or properties to be specified." + ) + + job_object = self._get_job(name) + if job_object.properties.job_type == RestJobType.PIPELINE: + job_object = self._get_job_2401(name) + if _is_pipeline_child_job(job_object): + raise PipelineChildJobError(job_id=job_object.id) + + from azure.ai.ml._restclient.runhistory.models import CreateRun + + self._runs_operations._operation.add_or_modify_by_experiment_name( + subscription_id=self._operation_scope.subscription_id, + resource_group_name=self._operation_scope.resource_group_name, + workspace_name=self._workspace_name, + experiment_name=job_object.properties.experiment_name, + run_id=name, + body=CreateRun( + display_name=display_name, + description=description, + tags=tags, + properties=properties, + ), + ) + + refreshed = self._get_job(name) + if refreshed.properties.job_type == RestJobType.PIPELINE: + refreshed = self._get_job_2401(name) + return self._resolve_azureml_id(Job._from_rest_object(refreshed)) + @distributed_trace @monitor_with_activity(ops_logger, "Job.Stream", ActivityType.PUBLICAPI) def stream(self, name: str) -> None: diff --git a/sdk/ml/azure-ai-ml/tests/job_common/unittests/test_job_operations.py b/sdk/ml/azure-ai-ml/tests/job_common/unittests/test_job_operations.py index 61eb51714df0..f3ba5ef64299 100644 --- a/sdk/ml/azure-ai-ml/tests/job_common/unittests/test_job_operations.py +++ b/sdk/ml/azure-ai-ml/tests/job_common/unittests/test_job_operations.py @@ -266,6 +266,168 @@ def test_restore(self, mock_method, mock_job_operation: JobOperations) -> None: mock_job_operation.service_client_01_2024_preview.jobs.get.assert_called_once() mock_job_operation._operation_2023_02_preview.create_or_update.assert_called_once() + # -------------- jobs.update() (new public API) -------------- + + def test_update_no_field_raises_user_error(self, mock_job_operation: JobOperations) -> None: + """update() with no updatable keyword must raise UserErrorException without any service calls.""" + from azure.ai.ml.exceptions import UserErrorException + + with pytest.raises(UserErrorException, match="at least one"): + mock_job_operation.update(name="random_name") + + @patch.object(Job, "_from_rest_object") + @patch.object(JobOperations, "_resolve_azureml_id") + @patch.object(JobOperations, "_get_job") + def test_update_routes_through_runhistory_patch( + self, mock_get_job, mock_resolve, mock_from_rest, mock_job_operation: JobOperations + ) -> None: + """update() with fields must invoke add_or_modify_by_experiment_name exactly once with + the correct experiment_name / run_id and a CreateRun body carrying the supplied fields. + The returned entity must also be routed through _resolve_azureml_id so callers get the + same resolved view they'd get from jobs.get().""" + from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType + + fake_job = Mock() + fake_job.properties.job_type = RestJobType.COMMAND + fake_job.properties.experiment_name = "exp1" + mock_get_job.return_value = fake_job + resolved_job = Command(component=None) + mock_from_rest.return_value = Command(component=None) + mock_resolve.return_value = resolved_job + # Force the property to return a fresh mock so add_or_modify_by_experiment_name is + # assertable (bypasses the lazy RunOperations construction that would fail on mocks). + mock_job_operation._runs_operations_client = Mock() + + result = mock_job_operation.update( + name="random_name", + display_name="new dn", + description="new desc", + tags={"k": "v"}, + properties={"pk": "pv"}, + ) + + add_mod = mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name + add_mod.assert_called_once() + call_kwargs = add_mod.call_args.kwargs + assert call_kwargs["experiment_name"] == "exp1" + assert call_kwargs["run_id"] == "random_name" + body = call_kwargs["body"] + assert body.display_name == "new dn" + assert body.description == "new desc" + assert body.tags == {"k": "v"} + assert body.properties == {"pk": "pv"} + mock_resolve.assert_called_once() + assert result is resolved_job + + @patch.object(Job, "_from_rest_object") + @patch.object(JobOperations, "_resolve_azureml_id") + @patch.object(JobOperations, "_get_job_2401") + @patch.object(JobOperations, "_get_job") + def test_update_pipeline_uses_2401_refetch( + self, + mock_get_job, + mock_get_job_2401, + mock_resolve, + mock_from_rest, + mock_job_operation: JobOperations, + ) -> None: + """PIPELINE jobs must be re-fetched via _get_job_2401 to obtain the non-projected view + before the RunHistory PATCH is issued. The refreshed entity is then resolved.""" + from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType + + pipeline_job = Mock() + pipeline_job.properties.job_type = RestJobType.PIPELINE + pipeline_job.properties.experiment_name = "exp1" + mock_get_job.return_value = pipeline_job + mock_get_job_2401.return_value = pipeline_job + mock_from_rest.return_value = Command(component=None) + mock_resolve.return_value = Command(component=None) + mock_job_operation._runs_operations_client = Mock() + + mock_job_operation.update(name="random_name", display_name="new dn") + + # _get_job_2401 is called at least once (initial resolve + refresh both hit it for + # PIPELINE jobs); the RunHistory PATCH must still be issued. + assert mock_get_job_2401.call_count >= 1 + mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name.assert_called_once() + mock_resolve.assert_called_once() + + @patch.object(JobOperations, "_get_job_2401") + @patch.object(JobOperations, "_get_job") + def test_update_pipeline_child_raises( + self, mock_get_job, mock_get_job_2401, mock_job_operation: JobOperations + ) -> None: + """A pipeline child job (properties is None on the 2401 view) must raise + PipelineChildJobError and must NOT issue any RunHistory PATCH.""" + from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType + from azure.ai.ml.exceptions import PipelineChildJobError + + parent_view = Mock() + parent_view.properties.job_type = RestJobType.PIPELINE + mock_get_job.return_value = parent_view + + child = Mock() + child.properties = None # _is_pipeline_child_job -> True + child.id = "/fake/child/id" + mock_get_job_2401.return_value = child + + mock_job_operation._runs_operations_client = Mock() + + with pytest.raises(PipelineChildJobError): + mock_job_operation.update(name="child_run", description="x") + + mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name.assert_not_called() + + # -------------- create_or_update metadata-only shortcut -------------- + + @patch.object(Job, "_from_rest_object") + @patch.object(JobOperations, "_resolve_azureml_id") + @patch.object(JobOperations, "_get_job") + def test_create_or_update_metadata_shortcut_uses_runhistory_patch( + self, + mock_get_job, + mock_resolve, + mock_from_rest, + mock_job_operation: JobOperations, + ) -> None: + """For a Job that was previously fetched (has an ARM id) and is being resubmitted with + only metadata edits (no compute/experiment_name change), create_or_update must route + through the RunHistory PATCH shortcut and skip the legacy MFE PUT round-trip.""" + from azure.ai.ml._restclient.v2023_08_01_preview.models import JobType as RestJobType + + fake_job = Mock() + fake_job.properties.job_type = RestJobType.COMMAND + fake_job.properties.experiment_name = "exp1" + mock_get_job.return_value = fake_job + mock_from_rest.return_value = Command(component=None) + mock_resolve.return_value = Command(component=None) + mock_job_operation._runs_operations_client = Mock() + + input_job = Mock() + input_job.id = ( + "/subscriptions/x/resourceGroups/y/providers/Microsoft.MachineLearningServices" + "/workspaces/z/jobs/random_name" + ) + input_job.name = "random_name" + input_job.display_name = "edited dn" + input_job.description = "edited desc" + input_job.tags = {"k": "v"} + input_job.properties = {"pk": "pv"} + input_job.compute = None + + mock_job_operation.create_or_update(job=input_job) + + add_mod = mock_job_operation._runs_operations._operation.add_or_modify_by_experiment_name + add_mod.assert_called_once() + body = add_mod.call_args.kwargs["body"] + assert body.display_name == "edited dn" + assert body.description == "edited desc" + assert body.tags == {"k": "v"} + assert body.properties == {"pk": "pv"} + # Legacy MFE PUT paths must be untouched when the shortcut succeeds. + mock_job_operation.service_client_01_2024_preview.jobs.create_or_update.assert_not_called() + mock_job_operation._operation_2023_02_preview.create_or_update.assert_not_called() + @pytest.mark.parametrize( "corrupt_job_data", [