Log created SageMaker resources and S3 locations

docs/tutorials/predictor-tabular.md

@@ -52,7 +52,7 @@ For class probabilities, use {py:meth}`~autogluon.cloud.TabularCloudPredictor.fi
 With `wait=False`, both methods return `None` right away; once {py:meth}`~autogluon.cloud.TabularCloudPredictor.get_fit_job_status` reports `Completed`, fetch the results with {py:meth}`~autogluon.cloud.TabularCloudPredictor.get_fit_predict_results` or {py:meth}`~autogluon.cloud.TabularCloudPredictor.get_fit_predict_proba_results`.
 
 ### Reattach to a training job
-If your local connection drops, the training job keeps running on SageMaker. You can reattach with another `CloudPredictor` via {py:meth}`~autogluon.cloud.TabularCloudPredictor.attach_job` as long as you have the job name — it's logged when training starts (``Start sagemaker training job `ag-cloud-tabular-...` ``) and also visible in the SageMaker console.
+If your local connection drops, the training job keeps running on SageMaker. You can reattach with another `CloudPredictor` via {py:meth}`~autogluon.cloud.TabularCloudPredictor.attach_job` as long as you have the job name — it's logged when training starts (``Creating SageMaker training job `ag-cloud-tabular-...` ``) and also visible in the SageMaker console.
 
 ```python
 another_cloud_predictor = TabularCloudPredictor()

docs/tutorials/predictor-timeseries.md

@@ -57,7 +57,7 @@ forecasts = cloud_predictor.fit_predict(
 By default predictions land at `{cloud_output_path}/{job_name}/predictions.csv`; pass `predictions_path` (ending in `.csv` or `.parquet`) to choose a destination. With `wait=False`, `fit_predict()` returns `None` right away; once {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.get_fit_job_status` reports `Completed`, fetch the forecast with {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.get_fit_predict_results`.
 
 ### Reattach to a training job
-If your local connection drops, the training job keeps running on SageMaker. You can reattach with another `CloudPredictor` via {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.attach_job` as long as you have the job name — it's logged when training starts (``Start sagemaker training job `ag-cloud-timeseries-...` ``) and also visible in the SageMaker console.
+If your local connection drops, the training job keeps running on SageMaker. You can reattach with another `CloudPredictor` via {py:meth}`~autogluon.cloud.TimeSeriesCloudPredictor.attach_job` as long as you have the job name — it's logged when training starts (``Creating SageMaker training job `ag-cloud-timeseries-...` ``) and also visible in the SageMaker console.
 
 ```python
 another_cloud_predictor = TimeSeriesCloudPredictor()

src/autogluon/cloud/backend/sagemaker_backend.py

@@ -25,7 +25,14 @@
     upload_training_code,
 )
 from ..utils.aws_utils import resolve_execution_role, setup_sagemaker_session
-from ..utils.constants import DEFAULT_FRAMEWORK_VERSION, DEFAULT_VOLUME_SIZE, LOCAL_MODE, LOCAL_MODE_GPU, VALID_ACCEPT
+from ..utils.constants import (
+    DEFAULT_FRAMEWORK_VERSION,
+    DEFAULT_VOLUME_SIZE,
+    LOCAL_MODE,
+    LOCAL_MODE_GPU,
+    MODEL_ARTIFACT_NAME,
+    VALID_ACCEPT,
+)
 from ..utils.deserializers import PandasDeserializer
 from ..utils.dlc_utils import infer_sagemaker_ami_version, parse_framework_version, retrieve_image_uri
 from ..utils.misc import MostRecentInsertedOrderedDict, sagemaker_timestamp, unique_name_from_base
@@ -362,8 +369,20 @@ def fit(
         }
         request = deep_merge(request, overrides.get("CreateTrainingJob", {}))
 
+        output_locations = {}
+        if ag_args.get("save_predictor", True):
+            model_output_path = request["OutputDataConfig"]["S3OutputPath"].rstrip("/")
+            output_locations["Predictor"] = f"{model_output_path}/{job_name}/output/{MODEL_ARTIFACT_NAME}"
+        if ag_args.get("predict_after_fit"):
+            output_locations["Predictions"] = ag_args["predictions_path"]
+
         self._fit_job = SageMakerFitJob(session=self.sagemaker_session)
-        self._fit_job.run(training_job_request=request, framework_version=framework_version, wait=wait)
+        self._fit_job.run(
+            training_job_request=request,
+            framework_version=framework_version,
+            wait=wait,
+            output_locations=output_locations,
+        )
 
     def _default_inference_framework_version(self, framework_version: str | None, uses_fit_output: bool) -> str:
         """Default to the version of the fit job when serving its model, so training and inference containers match."""
@@ -401,9 +420,7 @@ def _create_model(
             "Tags": tags,
         }
         request = deep_merge(request, overrides.get("CreateModel", {}))
-        logger.log(20, "Creating inference model...")
         self.sagemaker_session.sagemaker_client.create_model(**request)
-        logger.log(20, "Inference model created successfully")
         return request["ModelName"]
 
     def _prepare_model_data(
@@ -613,7 +630,8 @@ def deploy(
             overrides.get("CreateEndpoint", {}),
         )
 
-        logger.log(20, f"Deploying model to the endpoint (inference_mode={inference_mode})")
+        resources = f"instance_type={instance_type}" if inference_mode == "realtime" else "serverless"
+        logger.log(20, f"Creating SageMaker endpoint `{endpoint_request['EndpointName']}` ({resources})")
         client = self.sagemaker_session.sagemaker_client
         try:
             client.create_endpoint_config(**endpoint_config_request)
@@ -630,6 +648,7 @@ def deploy(
         self.endpoint_name = endpoint_request["EndpointName"]
         if wait:
             client.get_waiter("endpoint_in_service").wait(EndpointName=self.endpoint_name)
+            logger.log(20, f"Endpoint `{self.endpoint_name}` is in service")
 
     def _create_serve_script_tarball(
         self, serve_script_path: str, endpoint_name: str, requirements: list[str] | None = None
@@ -1056,8 +1075,9 @@ def _prepare_and_upload_data(self, data, filename, bucket, key_prefix, output_ty
         Returns the S3 URI of the uploaded file.
         """
         local_path = self._prepare_data(data, filename, output_type=output_type)
-        logger.log(20, f"Uploading {filename} data...")
-        return self.sagemaker_session.upload_data(path=local_path, bucket=bucket, key_prefix=key_prefix)
+        s3_uri = self.sagemaker_session.upload_data(path=local_path, bucket=bucket, key_prefix=key_prefix)
+        logger.log(20, f"Uploaded {filename} to {s3_uri}")
+        return s3_uri
 
     def _find_common_path_and_replace_image_column(self, data, image_column):
         common_path = os.path.commonpath(data[image_column].tolist())
@@ -1097,7 +1117,6 @@ def _upload_fit_artifact(
             inputs[channel_name] = self._prepare_and_upload_data(
                 channel_data, channel_name, cloud_bucket, util_key_prefix
             )
-            logger.log(20, f"{channel_name} uploaded successfully")
 
         inputs["ag_args"] = self.sagemaker_session.upload_data(
             path=ag_args, bucket=cloud_bucket, key_prefix=util_key_prefix
@@ -1134,13 +1153,12 @@ def _upload_fit_image_artifact(self, image_dir_path, bucket, key_prefix):
             logger.log(20, "Zipping images ...")
             zipfolder(image_zip_filename, image_dir_path)
             image_zip_filename += ".zip"
-            logger.log(20, "Uploading images ...")
             upload_image_path = self.sagemaker_session.upload_data(
                 path=image_zip_filename,
                 bucket=bucket,
                 key_prefix=key_prefix,
             )
-            logger.log(20, "Images uploaded successfully")
+            logger.log(20, f"Uploaded images to {upload_image_path}")
         return upload_image_path
 
     def _upload_predictor(self, predictor_path, key_prefix):
@@ -1151,6 +1169,7 @@ def _upload_predictor(self, predictor_path, key_prefix):
                     predictor_path = self.sagemaker_session.upload_data(
                         path=predictor_path, bucket=cloud_bucket, key_prefix=key_prefix
                     )
+                    logger.log(20, f"Uploaded predictor to {predictor_path}")
                 else:
                     raise ValueError("Please provide a tarball containing the model")
             else:
@@ -1202,9 +1221,8 @@ def _predict_real_time(self, test_data, accept, split_pred_proba=True, inference
     def _upload_batch_predict_data(self, test_data, bucket, key_prefix):
         if isinstance(test_data, pd.DataFrame):
             test_data = self._prepare_data(test_data, "test", output_type="csv")
-        logger.log(20, "Uploading data...")
         test_input = self.sagemaker_session.upload_data(path=test_data, bucket=bucket, key_prefix=key_prefix + "/data")
-        logger.log(20, "Data uploaded successfully")
+        logger.log(20, f"Uploaded test data to {test_input}")
 
         return test_input
 

src/autogluon/cloud/job/sagemaker_job.py

@@ -201,16 +201,24 @@ def run(
         training_job_request: dict[str, Any],
         framework_version: str | None,
         wait: bool,
+        output_locations: dict[str, str] | None = None,
     ):
-        """Create the training job from a ``CreateTrainingJob`` request and optionally wait for it to finish."""
+        """Create the training job from a ``CreateTrainingJob`` request and optionally wait for it to finish.
+
+        ``output_locations`` maps a description of each output (e.g. ``"Predictor"``) to the S3 location the job will
+        write it to; these are logged once the job is created.
+        """
         job_name = training_job_request["TrainingJobName"]
-        logger.log(20, f"Start sagemaker training job `{job_name}`")
+        logger.log(20, f"Creating SageMaker training job `{job_name}`")
         try:
             self.session.sagemaker_client.create_training_job(**training_job_request)
             self._job_name = job_name
+            for name, location in (output_locations or {}).items():
+                logger.log(20, f"{name} will be saved to {location}")
             self._framework_version = framework_version
             if wait:
                 self._wait_until_completed()
+                logger.log(20, f"Training job `{job_name}` completed")
         except Exception as e:
             logger.error(f"Training failed. Please check sagemaker console training jobs {job_name} for details.")
             raise e
@@ -260,23 +268,23 @@ def run(
         start. With ``wait=False`` the model is kept, since the job still needs it.
         """
         job_name = transform_job_request["TransformJobName"]
+        input_uri = transform_job_request["TransformInput"]["DataSource"]["S3DataSource"]["S3Uri"]
+        self._output_filename = input_uri.split("/")[-1] + ".out"
         try:
-            logger.log(20, "Transforming")
+            logger.log(20, f"Creating SageMaker batch transform job `{job_name}`")
             self.session.sagemaker_client.create_transform_job(**transform_job_request)
             self._job_name = job_name
+            output_path = transform_job_request["TransformOutput"]["S3OutputPath"] + "/" + self._output_filename
+            logger.log(20, f"Predictions will be saved to {output_path}")
             if wait:
                 self._wait_until_completed()
-            logger.log(20, "Transform done")
+                logger.log(20, f"Batch transform job `{job_name}` completed")
         except Exception as e:
             delete_quietly(self.session.sagemaker_client.delete_model, ModelName=model_name)
             raise e
 
-        input_uri = transform_job_request["TransformInput"]["DataSource"]["S3DataSource"]["S3Uri"]
-        self._output_filename = input_uri.split("/")[-1] + ".out"
-
         if wait:
             self._delete_model(model_name)
-            logger.log(20, f"Predict results have been saved to {self.get_output_path()}")
         else:
             logger.log(
                 20,