fix(train): raise on expired credentials in show_metrics log fetch

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.
         """

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()

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",
+    }
+)

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):