diff --git a/sagemaker-train/src/sagemaker/train/base_trainer.py b/sagemaker-train/src/sagemaker/train/base_trainer.py index 468dacd610..b32b2ada14 100644 --- a/sagemaker-train/src/sagemaker/train/base_trainer.py +++ b/sagemaker-train/src/sagemaker/train/base_trainer.py @@ -393,6 +393,8 @@ def show_metrics( Raises: NotImplementedError: If the training technique does not support metric extraction (e.g., DPO). + PermissionError: If CloudWatch logs cannot be read because the caller's + credentials are expired/invalid or lack CloudWatch Logs permissions. ValueError: If no training job has been run yet, no logs/metrics are found, or MLflow is not configured for OSS models. """ diff --git a/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py b/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py index ed1906675f..7a2ade9d43 100644 --- a/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py +++ b/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py @@ -15,8 +15,10 @@ import re from datetime import datetime from typing import Any, Dict, List, Optional +from botocore.exceptions import ClientError from sagemaker.core.training.configs import HyperPodCompute +from sagemaker.train.common_utils.constants import AUTH_ERROR_CODES logger = logging.getLogger(__name__) @@ -43,6 +45,32 @@ _UNSUPPORTED_TECHNIQUES = {"DPO", "RLAIF"} + +def _raise_if_auth_error(error: ClientError, job_name: str, operation: str) -> None: + """Re-raise a CloudWatch Logs error as ``PermissionError`` if it is auth-related. + + Return silently for any other error code so the caller can treat it as + "no log data available". + + Args: + error: The ``ClientError`` raised by the CloudWatch Logs client. + job_name: Training job name, included in the error message. + operation: CloudWatch Logs API that failed, e.g. "DescribeLogStreams". + + Raises: + PermissionError: If the error code indicates expired/invalid credentials + or missing CloudWatch Logs permissions. + """ + error_code = error.response.get("Error", {}).get("Code", "") + if error_code in AUTH_ERROR_CODES: + raise PermissionError( + f"Cannot read CloudWatch logs for job '{job_name}': {operation} failed with " + f"'{error_code}'. Verify your AWS credentials are valid and have not expired, " + f"and that your caller identity has logs:DescribeLogStreams, logs:GetLogEvents, " + f"and logs:FilterLogEvents permissions." + ) from error + + def _get_smtj_log_group() -> str: """Return the CW log group for SageMaker Training Jobs.""" return "/aws/sagemaker/TrainingJobs" @@ -67,14 +95,20 @@ def _fetch_smtj_logs( start_time: Optional[int] = None, end_time: Optional[int] = None, ) -> List[Dict[str, Any]]: - """Fetch CloudWatch log events for an SMTJ training job.""" + """Fetch CloudWatch log events for an SMTJ training job. + + Raises: + PermissionError: If the logs cannot be read because the caller's credentials + are expired/invalid or lack CloudWatch Logs permissions. + """ # Find the job's dedicated log stream try: response = logs_client.describe_log_streams( logGroupName=log_group, logStreamNamePrefix=job_name, ) - except Exception as e: + except ClientError as e: + _raise_if_auth_error(e, job_name, "DescribeLogStreams") logger.warning(f"Could not describe log streams for job '{job_name}': {e}") return [] @@ -102,7 +136,12 @@ def _fetch_smtj_logs( if next_token: params["nextToken"] = next_token - response = logs_client.get_log_events(**params) + try: + response = logs_client.get_log_events(**params) + except ClientError as e: + _raise_if_auth_error(e, job_name, "GetLogEvents") + raise + events = response.get("events", []) all_events.extend(events) @@ -125,6 +164,10 @@ def _fetch_smhp_logs( HyperPod doesn't separate log streams by job — uses filter_log_events with the job ID as the filter pattern. + + Raises: + PermissionError: If the logs cannot be read because the caller's credentials + are expired/invalid or lack CloudWatch Logs permissions. """ all_events: List[Dict[str, Any]] = [] next_token = None @@ -144,7 +187,8 @@ def _fetch_smhp_logs( try: response = logs_client.filter_log_events(**params) - except Exception as e: + except ClientError as e: + _raise_if_auth_error(e, job_id, "FilterLogEvents") logger.warning(f"Could not filter log events for HP job '{job_id}': {e}") return all_events @@ -346,6 +390,8 @@ def fetch_and_plot_metrics( Raises: NotImplementedError: If the technique is not supported (e.g., DPO). + PermissionError: If CloudWatch logs cannot be read because the caller's + credentials are expired/invalid or lack CloudWatch Logs permissions. ValueError: If no logs or metrics are found. """ technique = customization_technique.upper() diff --git a/sagemaker-train/src/sagemaker/train/common_utils/constants.py b/sagemaker-train/src/sagemaker/train/common_utils/constants.py index 36966f5a37..85deef19ca 100644 --- a/sagemaker-train/src/sagemaker/train/common_utils/constants.py +++ b/sagemaker-train/src/sagemaker/train/common_utils/constants.py @@ -113,3 +113,19 @@ class _ErrorConstants: # Minimum MLflow version required for MTRL training MIN_MLFLOW_VERSION = "3.10" + +# Error codes that mean "we could not read the resource" rather than "the resource +# is empty". These must surface to the caller instead of degrading to an empty +# result, otherwise the user is told their job has no logs when their credentials +# are the real problem. +AUTH_ERROR_CODES = frozenset( + { + "AccessDenied", + "AccessDeniedException", + "ExpiredToken", + "ExpiredTokenException", + "InvalidClientTokenId", + "UnauthorizedOperation", + "UnrecognizedClientException", + } +) diff --git a/sagemaker-train/tests/unit/train/common_utils/test_cloudwatch_metrics.py b/sagemaker-train/tests/unit/train/common_utils/test_cloudwatch_metrics.py index a67fd31c1d..5edcb40a29 100644 --- a/sagemaker-train/tests/unit/train/common_utils/test_cloudwatch_metrics.py +++ b/sagemaker-train/tests/unit/train/common_utils/test_cloudwatch_metrics.py @@ -7,6 +7,7 @@ import pandas import pytest +from botocore.exceptions import ClientError from sagemaker.core.training.configs import Compute, HyperPodCompute from sagemaker.train.base_trainer import BaseTrainer @@ -16,6 +17,7 @@ fetch_and_plot_metrics, parse_metrics_from_logs, ) +from sagemaker.train.common_utils.constants import AUTH_ERROR_CODES FAKE_SFT_LOGS = [ @@ -37,6 +39,14 @@ ] +def _client_error(code, operation): + """Build a botocore ClientError with the given error code.""" + return ClientError( + {"Error": {"Code": code, "Message": f"simulated {code}"}}, + operation, + ) + + class TestParseMetrics: def test_sft_extracts_loss_and_lr(self): @@ -121,6 +131,75 @@ def test_smhp_uses_filter_with_job_id(self): assert '"hp-job-123"' in call_kwargs["filterPattern"] +class TestFetchLogsAuthErrors: + """Credential/permission failures must surface, not degrade to an empty log list.""" + + @pytest.mark.parametrize("error_code", sorted(AUTH_ERROR_CODES)) + def test_smtj_describe_streams_auth_error_raises(self, error_code): + mock_client = MagicMock() + mock_client.describe_log_streams.side_effect = _client_error( + error_code, "DescribeLogStreams" + ) + + with pytest.raises(PermissionError, match="credentials"): + _fetch_smtj_logs("my-job", mock_client, "/aws/sagemaker/TrainingJobs") + + def test_smtj_expired_token_message_has_job_and_code(self): + mock_client = MagicMock() + mock_client.describe_log_streams.side_effect = _client_error( + "ExpiredTokenException", "DescribeLogStreams" + ) + + with pytest.raises(PermissionError) as exc_info: + _fetch_smtj_logs("my-job", mock_client, "/aws/sagemaker/TrainingJobs") + + message = str(exc_info.value) + assert "my-job" in message + assert "ExpiredTokenException" in message + + def test_smtj_missing_log_group_returns_empty(self): + """A genuinely absent log group is not an auth failure — degrade to [].""" + mock_client = MagicMock() + mock_client.describe_log_streams.side_effect = _client_error( + "ResourceNotFoundException", "DescribeLogStreams" + ) + + events = _fetch_smtj_logs("my-job", mock_client, "/aws/sagemaker/TrainingJobs") + assert events == [] + + def test_smtj_get_log_events_auth_error_raises(self): + """Auth failure part-way through pagination surfaces as PermissionError.""" + mock_client = MagicMock() + mock_client.describe_log_streams.return_value = { + "logStreams": [{"logStreamName": "my-job/algo-1"}] + } + mock_client.get_log_events.side_effect = _client_error( + "ExpiredTokenException", "GetLogEvents" + ) + + with pytest.raises(PermissionError, match="credentials"): + _fetch_smtj_logs("my-job", mock_client, "/aws/sagemaker/TrainingJobs") + + @pytest.mark.parametrize("error_code", sorted(AUTH_ERROR_CODES)) + def test_smhp_filter_events_auth_error_raises(self, error_code): + mock_client = MagicMock() + mock_client.filter_log_events.side_effect = _client_error( + error_code, "FilterLogEvents" + ) + + with pytest.raises(PermissionError, match="credentials"): + _fetch_smhp_logs("hp-job-123", mock_client, "/aws/sagemaker/Clusters/c/id") + + def test_smhp_missing_log_group_returns_empty(self): + mock_client = MagicMock() + mock_client.filter_log_events.side_effect = _client_error( + "ResourceNotFoundException", "FilterLogEvents" + ) + + events = _fetch_smhp_logs("hp-job-123", mock_client, "/aws/sagemaker/Clusters/c/id") + assert events == [] + + class TestFetchAndPlotMetrics: def _session(self): @@ -174,6 +253,32 @@ def test_no_logs_found_raises(self, mock_fetch): "SFT", self._session(), ) + def test_job_without_logs_yet_still_raises_no_logs_found(self): + """A just-started job with no log stream keeps reporting "No CloudWatch logs found".""" + session = self._session() + session.boto_session.client.return_value.describe_log_streams.return_value = { + "logStreams": [] + } + + with pytest.raises(ValueError, match="No CloudWatch logs found"): + fetch_and_plot_metrics( + "just-started-job", Compute(instance_type="ml.p5.48xlarge", instance_count=1), + "SFT", session, + ) + + def test_expired_credentials_raises_instead_of_no_logs_found(self): + """Expired credentials must not be reported as a job with no logs.""" + session = self._session() + session.boto_session.client.return_value.describe_log_streams.side_effect = ( + _client_error("ExpiredTokenException", "DescribeLogStreams") + ) + + with pytest.raises(PermissionError, match="credentials"): + fetch_and_plot_metrics( + "my-job", Compute(instance_type="ml.p5.48xlarge", instance_count=1), + "SFT", session, + ) + @patch("sagemaker.train.common_utils.cloudwatch_metrics.plot_metrics") @patch("sagemaker.train.common_utils.cloudwatch_metrics._fetch_smtj_logs") def test_result_sorted_by_step(self, mock_fetch, mock_plot):