fix: tail_lines in stream_logs returns last N events (true tail seman…

[amazeAmazing]

Issue #, if available:
N/A

Description of changes:

Issue

stream_logs(tail_lines=N) prints the first N log lines (head behavior) instead of the last N (tail behavior). Users expect tail_lines to work like tail -n or kubectl logs --tail — showing the most recent output.

Description of changes

Bug fix: true tail semantics for tail_lines

When tail_lines is set, stream_logs() now fetches the last N log events from CloudWatch and returns immediately.

SMTJ path (get_log_events): Uses backward pagination via nextBackwardToken to read from the end of each log stream. For multi-stream jobs (distributed training), merges events across all streams by timestamp and returns the globally last N.

SMHP path (filter_log_events): Uses startFromHead=False with pagination. CloudWatch bounds pages by bytes scanned (not matches found), so we follow nextToken until N matching events are collected.

Refactor: eliminate duplicated SMHP inline loop

_stream_logs_smhp previously had ~80 lines of inline polling logic that duplicated what LogStreamer + stream_log_loop already provide. Replaced the inline loop with a LogStreamer instance in filter mode, delegating to stream_log_loop for consistent behavior across all paths (SMTJ, SMHP, MTRL).

Simplify _print_event and remove dead counter

Previously, _print_event had a dual role: print the log line AND check a lines_printed counter to stop at tail_lines. Callers used if _print_event(...): return to exit the loop when the limit was reached. This was the old "head" behavior (stream forward, stop after N).

Now that tail_lines is handled before the loop via poll_tail() (which fetches the last N events and returns immediately), the streaming loop only runs when tail_lines is None. The counter and conditional return are dead code — they can never trigger. _print_event is now a simple print helper with no return value.

The limit is enforced inside poll_tail():

• SMTJ (_tail_stream_mode): passes limit=N to get_log_events per stream, then merges across streams and slices to the globally last N with all_results[-n:].
• SMHP (_tail_filter_mode): accumulates events from paginated filter_log_events calls and stops when len(results) >= n, then slices to results[:n] and reverses to chronological order.

Validation

_tail_filter_mode raises ValueError if start_time is before 2024-01-01 — CloudWatch requires this for startFromHead=False on filter_log_events.

Testing

• 8 new unit tests covering poll_tail (stream mode, filter mode, multi-stream merge, multi-page pagination, pre-2024 validation, stream_log_loop integration)
• All 29 unit tests in test_log_streamer.py pass
• Manually tested against CloudWatch logs in us-east-1 in separate AWS account

Files changed

File	Change
sagemaker-train/src/sagemaker/train/common_utils/log_streamer.py	Add poll_tail(), _tail_stream_mode(), _tail_filter_mode(). Update stream_log_loop to use poll_tail when tail_lines is set.
sagemaker-train/src/sagemaker/train/base_trainer.py	Replace _stream_logs_smhp inline loop with LogStreamer + stream_log_loop. Remove unused imports (os, ClientError, MultiLogStreamHandler).
sagemaker-train/tests/unit/train/test_log_streamer.py	Add 8 unit tests for tail_lines behavior.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.