[ZealSV]

Issue #, if available:

Description of changes:

Adds describe_endpoint_traffic() and Workload.from_endpoint() to sagemaker.serve.ai_inference_recommender, so a benchmark or recommendation job can be sized from the traffic an endpoint serves today.

describe_endpoint_traffic(endpoint_name, *, inference_component_name=None, variant_name=None, lookback=None, start_time=None, end_time=None, sagemaker_session=None) returns an EndpointTraffic for one model on the endpoint:

Field	What it is
concurrency	Concurrent requests on the whole deployment: each hour's maximum ConcurrentRequestsPerCopy (inference component) or ConcurrentRequestsPerModel (production variant), times the current copies or instances. max is the busiest hour's.
requests, input_tokens, output_tokens	The requests the model completed in the window, and their prompt and generated tokens, from the serving container's token histograms.
framework	The serving framework whose histograms were read ("vllm"), or None.
detailed_observability	The endpoint configuration's MetricsConfig.EnableDetailedObservability. The token histograms aren't read when it's False.
data_capture_s3_uri	The variant's Data Capture prefix, when the endpoint captures requests.
start_time, end_time	The window, in UTC.

concurrency, input_tokens and output_tokens are TrafficStats: min, p25, p50, p70, p80, p90, p95, p99, max (p100), mean and stddev.

• Window: relative (lookback back from end_time, seven days by default) or absolute (start_time to end_time), up to seven days. Times must be timezone-aware; naive datetimes are refused.
• Token metrics: with detailed observability on, the serving container's per-request token histograms are published to CloudWatch. They're read through CloudWatch's Prometheus-compatible query API, signed with the session's credentials (cloudwatch:GetMetricData, cloudwatch:ListMetrics). The histograms are listed per framework (today vLLM's vllm:request_prompt_tokens and vllm:request_generation_tokens), so adding a framework is one entry.
• Method: the histograms are cumulative, so a series' requests in the window are its newest snapshot in the window less its snapshot at the start, read with one instant query per day of the window. A series that started in the window, or restarted, counts from zero. The request count and the token means are exact. The percentiles interpolate within the buckets, and the standard deviation uses the bucket midpoints.
• Adapters: an adapter's requests run in its base component's container, which publishes their concurrency and token histograms. So passing an adapter describes its base component, and finding the component skips adapters.

Workload.from_endpoint(endpoint_name, ...) replays the variant's captured requests (from_dataset(..., custom_dataset_type="sagemaker-datacapture")) when the endpoint has Data Capture on. Otherwise it builds a synthetic workload from the token metrics' means and standard deviations. Either way, the concurrency is the busiest hour's unless concurrency is passed. use_data_capture=False opts out of replaying captured requests, which can contain end users' data. Other workload parameters (tokenizer, hf_token, request_count, streaming, ...) pass through.

from datetime import timedelta
from sagemaker.serve.ai_inference_recommender import Workload, describe_endpoint_traffic

traffic = describe_endpoint_traffic("my-endpoint", inference_component_name="my-ic", lookback=timedelta(days=3))
print(traffic.concurrency.max, traffic.input_tokens.p50, traffic.output_tokens.p99)

workload = Workload.from_endpoint("my-endpoint", inference_component_name="my-ic", tokenizer="Qwen/Qwen3-0.6B")

Testing done:

• Unit tests: sagemaker-serve/tests/unit/test_ai_inference_recommender/test_traffic.py (61 tests, 100% line coverage of traffic.py). The test_ai_inference_recommender suite passes (284 tests).
• black, flake8, pylint (10.00/10) and pydocstyle pass on the changed files.
• Read-only runs against live endpoints in a test account. On an inference component under benchmark load, the request count matched the component's Invocations exactly (169, then 401 requests) on windows that ended a few minutes earlier. Runs also covered a LoRA adapter, components and variants without traffic, and an endpoint whose endpoint configuration was deleted while it was in service.

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

[github-actions]

🤖 AI code review

This PR adds describe_endpoint_traffic() and Workload.from_endpoint() to sagemaker.serve.ai_inference_recommender, deriving concurrency from CloudWatch invocation metrics and token-size distributions from the serving container's token histograms read through CloudWatch's Prometheus-compatible query API. The implementation is well-structured, follows the module's established conventions (session resolution via _boto_session, @_telemetry_emitter with Feature.INFERENCE_RECOMMENDER), and is backed by an extensive, well-targeted unit test suite covering windowing, histogram math, target resolution, data capture, query-API errors, and from_endpoint paths. I found only minor robustness/doc nits; no correctness or breaking-change issues.

_{Reviewed commit f234e56. Automated review; verify before acting.}

[github-actions]

🤖 AI code review

This PR adds describe_endpoint_traffic() and Workload.from_endpoint() to sagemaker.serve.ai_inference_recommender, letting a benchmark/recommendation workload be sized from an endpoint's live traffic. It reads concurrency from CloudWatch GetMetricStatistics and token-size histograms through CloudWatch's Prometheus-compatible query API (SigV4-signed GETs), estimating percentiles/stddev from histogram buckets. The implementation follows established module conventions (_boto_session reuse, @_telemetry_emitter with Feature.INFERENCE_RECOMMENDER as in listing.py) and is backed by a large, well-targeted unit test suite plus integration coverage. Overall the code is clean and correct; I only have minor observations.

_{Reviewed commit 7f0f2fd. Automated review; verify before acting.}

[github-actions]

🟡 Low: _PromQLClient reaches into botocore internals (cloudwatch.meta.endpoint_url and cloudwatch._endpoint.http_session). The latter is a private attribute and is not part of botocore's public contract, so it can break silently on a botocore upgrade. This is a known risk given the Prometheus-query API isn't exposed by boto3, but it would be worth pinning a minimum botocore version or adding a focused test that fails loudly if http_session disappears, so the breakage surfaces at test time rather than at runtime.

[rsareddy0329]

The sagemaker-mlops integ test failure is unrelated to this change - it's caused by a JS change and will be fixed in #6553.
Merging this PR since the failure isn't from it.