From 7853320a874c8818d7014c69f396b60073f87212 Mon Sep 17 00:00:00 2001 From: sayemkam Date: Mon, 27 Jul 2026 19:41:40 +0000 Subject: [PATCH 1/2] fix(train): raise on expired credentials in show_metrics log fetch When AWS credentials expire, show_metrics() reported "No CloudWatch logs found for job ''. The job may still be starting, or logs may not be available yet", sending users to debug their training job instead of refreshing credentials. _fetch_smtj_logs() and _fetch_smhp_logs() each wrapped their CloudWatch Logs call in a bare `except Exception`, logged a warning, and returned an empty list, so an ExpiredTokenException was indistinguishable from a job that genuinely has no logs yet. The empty list then became the misleading ValueError above. Design follows the existing notifications.py pattern: read the structured error code, re-raise the auth subset as PermissionError with remediation guidance, and let every other code keep its current degrade-to-empty behavior. A genuinely absent log group (ResourceNotFoundException) still returns an empty list, so a just-started job continues to raise the existing "No CloudWatch logs found" ValueError. Changes: - cloudwatch_metrics.py: add _AUTH_ERROR_CODES and _raise_if_auth_error(); apply at describe_log_streams, get_log_events, and filter_log_events; narrow `except Exception` to `except ClientError` - base_trainer.py: document PermissionError on show_metrics() - test_cloudwatch_metrics.py: 20 unit tests covering all 7 auth codes on both platforms, ResourceNotFoundException degradation, mid-pagination failure, and a regression guard for the existing "no logs found" path --- .../src/sagemaker/train/base_trainer.py | 2 + .../train/common_utils/cloudwatch_metrics.py | 66 +++++++++- .../common_utils/test_cloudwatch_metrics.py | 114 ++++++++++++++++++ 3 files changed, 178 insertions(+), 4 deletions(-) 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..36b5d6f619 100644 --- a/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py +++ b/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py @@ -15,6 +15,7 @@ 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 @@ -43,6 +44,45 @@ _UNSUPPORTED_TECHNIQUES = {"DPO", "RLAIF"} +# Error codes that mean "we could not read the logs" rather than "there are no logs". +# 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 = { + "AccessDenied", + "AccessDeniedException", + "ExpiredToken", + "ExpiredTokenException", + "InvalidClientTokenId", + "UnauthorizedOperation", + "UnrecognizedClientException", +} + + +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 +107,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 +148,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 +176,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 +199,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 +402,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/tests/unit/train/common_utils/test_cloudwatch_metrics.py b/sagemaker-train/tests/unit/train/common_utils/test_cloudwatch_metrics.py index a67fd31c1d..57a779ec9a 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 @@ -36,6 +37,24 @@ {"message": "global_step: 2 train_rm_score: 0.72"}, ] +AUTH_ERROR_CODES = [ + "AccessDenied", + "AccessDeniedException", + "ExpiredToken", + "ExpiredTokenException", + "InvalidClientTokenId", + "UnauthorizedOperation", + "UnrecognizedClientException", +] + + +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: @@ -121,6 +140,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", 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", 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 +262,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): From 6e2b58e1231f6d33e72bd594f1f2ec2168799312 Mon Sep 17 00:00:00 2001 From: sayemkam Date: Tue, 28 Jul 2026 18:26:17 +0000 Subject: [PATCH 2/2] change(train): move AUTH_ERROR_CODES to common_utils/constants.py Addresses review feedback: the auth error-code list was duplicated in cloudwatch_metrics.py and its unit test, so the two lists had to be kept in sync manually. Move it to the existing common_utils/constants.py as a module-level frozenset and import it in both places. Test parametrization sorts the frozenset so test IDs stay deterministic. --- .../train/common_utils/cloudwatch_metrics.py | 16 ++-------------- .../sagemaker/train/common_utils/constants.py | 16 ++++++++++++++++ .../common_utils/test_cloudwatch_metrics.py | 15 +++------------ 3 files changed, 21 insertions(+), 26 deletions(-) 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 36b5d6f619..7a2ade9d43 100644 --- a/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py +++ b/sagemaker-train/src/sagemaker/train/common_utils/cloudwatch_metrics.py @@ -18,6 +18,7 @@ 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__) @@ -44,19 +45,6 @@ _UNSUPPORTED_TECHNIQUES = {"DPO", "RLAIF"} -# Error codes that mean "we could not read the logs" rather than "there are no logs". -# 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 = { - "AccessDenied", - "AccessDeniedException", - "ExpiredToken", - "ExpiredTokenException", - "InvalidClientTokenId", - "UnauthorizedOperation", - "UnrecognizedClientException", -} - 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. @@ -74,7 +62,7 @@ def _raise_if_auth_error(error: ClientError, job_name: str, operation: str) -> N or missing CloudWatch Logs permissions. """ error_code = error.response.get("Error", {}).get("Code", "") - if error_code in _AUTH_ERROR_CODES: + 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, " 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 57a779ec9a..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 @@ -17,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,16 +38,6 @@ {"message": "global_step: 2 train_rm_score: 0.72"}, ] -AUTH_ERROR_CODES = [ - "AccessDenied", - "AccessDeniedException", - "ExpiredToken", - "ExpiredTokenException", - "InvalidClientTokenId", - "UnauthorizedOperation", - "UnrecognizedClientException", -] - def _client_error(code, operation): """Build a botocore ClientError with the given error code.""" @@ -143,7 +134,7 @@ def test_smhp_uses_filter_with_job_id(self): class TestFetchLogsAuthErrors: """Credential/permission failures must surface, not degrade to an empty log list.""" - @pytest.mark.parametrize("error_code", AUTH_ERROR_CODES) + @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( @@ -189,7 +180,7 @@ def test_smtj_get_log_events_auth_error_raises(self): with pytest.raises(PermissionError, match="credentials"): _fetch_smtj_logs("my-job", mock_client, "/aws/sagemaker/TrainingJobs") - @pytest.mark.parametrize("error_code", AUTH_ERROR_CODES) + @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(