# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
"""Telemetry module for SageMaker Python SDK to collect usage data and metrics."""
from __future__ import absolute_import
import logging
import os
import platform
import sys
from time import perf_counter
from typing import List
import functools
import requests
from urllib.parse import quote
import boto3
from botocore.exceptions import (
ParamValidationError,
NoCredentialsError,
PartialCredentialsError,
ConnectTimeoutError,
ReadTimeoutError,
EndpointConnectionError,
ConnectionClosedError,
ClientError,
NoRegionError,
)
from sagemaker.core.apiutils._boto_functions import to_lower_camel_case
from sagemaker.core.helper.session_helper import Session
from sagemaker.core.telemetry.attribution import _CREATED_BY_ENV_VAR
from sagemaker.core.telemetry.resource_creation import get_resource_arn
from sagemaker.core.common_utils import resolve_value_from_config
from sagemaker.core.config.config_schema import TELEMETRY_OPT_OUT_PATH
from sagemaker.core.telemetry.constants import (
Feature,
Status,
Region,
DEFAULT_AWS_REGION,
)
from sagemaker.core.user_agent import SDK_VERSION, process_studio_metadata_file
logger = logging.getLogger(__name__)
OS_NAME = platform.system() or "UnresolvedOS"
OS_VERSION = platform.release() or "UnresolvedOSVersion"
OS_NAME_VERSION = "{}/{}".format(OS_NAME, OS_VERSION)
PYTHON_VERSION = "{}.{}.{}".format(
sys.version_info.major, sys.version_info.minor, sys.version_info.micro
)
TELEMETRY_OPT_OUT_MESSAGING = (
"SageMaker Python SDK will collect telemetry to help us better understand our user's needs, "
"diagnose issues, and deliver additional features.\n"
"To opt out of telemetry, please disable via TelemetryOptOut parameter in SDK defaults config. "
"For more information, refer to https://sagemaker.readthedocs.io/en/stable/overview.html"
"#configuring-and-using-defaults-with-the-sagemaker-python-sdk."
)
FEATURE_TO_CODE = {
str(Feature.SDK_DEFAULTS): 11,
str(Feature.LOCAL_MODE): 12,
str(Feature.REMOTE_FUNCTION): 13,
str(Feature.MODEL_TRAINER): 14,
str(Feature.MODEL_CUSTOMIZATION): 15,
str(Feature.MLOPS): 16,
str(Feature.FEATURE_STORE): 17,
str(Feature.PROCESSING): 18,
str(Feature.MODEL_CUSTOMIZATION_NOVA): 19,
str(Feature.MODEL_CUSTOMIZATION_OSS): 20,
}
STATUS_TO_CODE = {
str(Status.SUCCESS): 1,
str(Status.FAILURE): 0,
}
# Exception type to error category mapping
# Botocore exceptions: https://github.com/boto/botocore/blob/develop/botocore/exceptions.py
# Python built-in exceptions: https://docs.python.org/3/library/exceptions.html
_EXCEPTION_TYPE_MAP = {
"validation_error": (ParamValidationError, NoRegionError, ValueError, TypeError),
"auth_error": (NoCredentialsError, PartialCredentialsError),
"timeout_error": (ConnectTimeoutError, ReadTimeoutError, TimeoutError),
"network_error": (EndpointConnectionError, ConnectionClosedError, ConnectionError, OSError),
}
# HTTP status code to error category mapping
# Reference: https://docs.aws.amazon.com/boto3/latest/guide/error-handling.html
_HTTP_STATUS_MAP = {
400: "validation_error",
401: "auth_error",
403: "auth_error",
404: "resource_not_found",
408: "timeout_error",
429: "throttling_error",
}
def _classify_error(e: Exception) -> str:
"""Classify an exception into an actionable error category.
Classification priority:
1. Exception type matching (botocore + Python built-ins)
2. HTTP status code from AWS service response
3. Fallback to exception class name
"""
# 1. Classify by exception type
# Botocore static exceptions: https://github.com/boto/botocore/blob/develop/botocore/exceptions.py
# Python built-in exceptions: https://docs.python.org/3/library/exceptions.html
for category, exception_types in _EXCEPTION_TYPE_MAP.items():
if isinstance(e, exception_types):
return category
# 2. Classify by HTTP status code from AWS service response
# Reference: https://docs.aws.amazon.com/boto3/latest/guide/error-handling.html
http_status = 0
if hasattr(e, "response") and isinstance(e.response, dict):
http_status = e.response.get("ResponseMetadata", {}).get("HTTPStatusCode", 0)
if http_status in _HTTP_STATUS_MAP:
return _HTTP_STATUS_MAP[http_status]
if 500 <= http_status < 600:
return "service_error"
return e.__class__.__name__.lower()
[docs]
class TelemetryParamType:
"""Constants for telemetry parameter extraction types.
Used in the `telemetry_params` list passed to @_telemetry_emitter decorator.
Each entry in telemetry_params is a tuple of (name, type) or (name, type, value).
To add a new telemetry signal to any class:
1. Identify what you want to track (instance attribute, method return, or kwarg).
2. Pick the appropriate type constant below.
3. Add a tuple to the `telemetry_params` list on the decorator.
Example:
@_telemetry_emitter(
feature=Feature.MODEL_CUSTOMIZATION,
func_name="MyClass.my_method",
telemetry_params=[
("model_name", TelemetryParamType.ATTR_VALUE), # emits x-modelName=<value>
("networking", TelemetryParamType.ATTR_EXISTS), # emits x-hasNetworking=true/false
("_is_fine_tuned", TelemetryParamType.ATTR_CALL), # emits x-isFineTuned=True/False
("instance_type", TelemetryParamType.KWARG_VALUE), # emits x-instanceType=<kwarg value>
("kms_key_id", TelemetryParamType.KWARG_EXISTS), # emits x-hasKmsKeyId=true/false
],
)
"""
# Reads self.<name> and emits the actual value.
# Use for: model names, training types, modes — values useful for analytics.
# Emits nothing if the attribute is None.
ATTR_VALUE = "attr_value"
# Reads self.<name> and emits true/false based on whether it's set (not None).
# Use for: sensitive configs (KMS, VPC, MLflow) where you only need to know
# if the customer configured it, without exposing the actual value.
ATTR_EXISTS = "attr_exists"
# Calls self.<name>() and emits the return value.
# Use for: computed/derived values like _is_model_customization(), _is_nova_model().
ATTR_CALL = "attr_call"
# Reads kwargs[<name>] from the decorated method's keyword arguments and emits the value.
# Use for: method parameters not stored on self (e.g., instance_type passed to deploy()).
# Emits nothing if the kwarg is None or not provided.
KWARG_VALUE = "kwarg_value"
# Reads kwargs[<name>] and emits true/false based on whether it's provided and truthy.
# Use for: optional method parameters where you only need presence info
# (e.g., update_endpoint, imported_model_kms_key_id).
KWARG_EXISTS = "kwarg_exists"
# Reads type(self.<name>).__name__ and emits the class name of the attribute.
# Use for: polymorphic attributes where the subclass determines the code path
# (e.g., compute → "HyperPodCompute"/"TrainingJobCompute", distributed → "Torchrun"/"MPI").
# Emits nothing if the attribute is None.
ATTR_TYPE = "attr_type"
def _extract_telemetry_params(instance, kwargs, telemetry_params=None) -> str:
"""Extract telemetry params from instance/kwargs based on telemetry_params list.
Args:
instance: The class instance (args[0]).
kwargs: The kwargs dict from the decorated function call.
telemetry_params: List of tuples defining what to extract.
- ("attr_name", ATTR_VALUE) → emit self.attr value
- ("attr_name", ATTR_EXISTS) → emit true/false
- ("method_name", ATTR_CALL) → call self.method(), emit return value
- ("kwarg_name", KWARG_VALUE) → emit kwargs value
- ("kwarg_name", KWARG_EXISTS) → emit true/false
- ("attr_name", ATTR_TYPE) → emit type(self.attr).__name__
Returns:
str: URL query params string.
"""
if not telemetry_params:
return ""
parts = []
T = TelemetryParamType
for param in telemetry_params:
name, kind = param[0], param[1]
key = to_lower_camel_case(name.lstrip("_"))
if kind == T.ATTR_VALUE:
value = getattr(instance, name, None)
if value is not None:
parts.append(f"&x-{key}={value}")
elif kind == T.ATTR_EXISTS:
value = getattr(instance, name, None)
parts.append(f"&x-has{key[0].upper()}{key[1:]}={'true' if value else 'false'}")
elif kind == T.ATTR_CALL:
method = getattr(instance, name, None)
if callable(method):
try:
parts.append(f"&x-{key}={method()}")
except Exception:
pass
elif kind == T.KWARG_VALUE:
value = kwargs.get(name) if kwargs else None
if value is not None:
parts.append(f"&x-{key}={value}")
elif kind == T.KWARG_EXISTS:
value = kwargs.get(name) if kwargs else None
parts.append(f"&x-has{key[0].upper()}{key[1:]}={'true' if value else 'false'}")
elif kind == T.ATTR_TYPE:
value = getattr(instance, name, None)
if value is not None:
parts.append(f"&x-{key}Type={type(value).__name__}")
return "".join(parts)
def _telemetry_emitter(feature: str, func_name: str, telemetry_params=None):
"""Telemetry Emitter
Decorator to emit telemetry logs for SageMaker Python SDK functions. This class needs
sagemaker_session object as a member. Default session object is a pysdk v2 Session object
in this repo. When collecting telemetry for classes using sagemaker-core Session object,
we should be aware of its differences, such as sagemaker_session.sagemaker_config does not
exist in new Session class.
Args:
feature: The Feature enum value for this telemetry event.
func_name: Human-readable function name for tracking.
telemetry_params: Optional list of tuples defining granular params to extract.
See TelemetryParamType for available types.
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
sagemaker_session = None
if len(args) > 0 and hasattr(args[0], "sagemaker_session"):
# Get the sagemaker_session from the instance method args
sagemaker_session = args[0].sagemaker_session
elif len(args) > 0 and hasattr(args[0], "_sagemaker_session"):
# Get the sagemaker_session from the instance method args (private attribute)
sagemaker_session = args[0]._sagemaker_session
elif feature == Feature.REMOTE_FUNCTION:
# Get the sagemaker_session from the function keyword arguments for remote function
sagemaker_session = kwargs.get(
"sagemaker_session", _get_default_sagemaker_session()
)
# Fallback: check kwargs for sagemaker_session (e.g., classmethods where
# args[0] is the class and the session is passed as a keyword argument)
if not sagemaker_session:
sagemaker_session = kwargs.get("sagemaker_session") or (
_get_default_sagemaker_session()
)
if sagemaker_session:
logger.debug("sagemaker_session found, preparing to emit telemetry...")
logger.info(TELEMETRY_OPT_OUT_MESSAGING)
response = None
caught_ex = None
studio_app_type = process_studio_metadata_file()
# Check if telemetry is opted out
telemetry_opt_out_flag = resolve_value_from_config(
direct_input=None,
config_path=TELEMETRY_OPT_OUT_PATH,
default_value=False,
sagemaker_session=sagemaker_session,
)
logger.debug("TelemetryOptOut flag is set to: %s", telemetry_opt_out_flag)
# Construct the feature list to track feature combinations
feature_list: List[int] = [FEATURE_TO_CODE[str(feature)]]
# For MODEL_CUSTOMIZATION, append NOVA or OSS sub-feature
# based on the instance's _is_nova_model_for_telemetry() method
if feature == Feature.MODEL_CUSTOMIZATION and len(args) > 0:
instance = args[0]
try:
if hasattr(instance, "_is_nova_model_for_telemetry"):
if instance._is_nova_model_for_telemetry():
feature_list.append(
FEATURE_TO_CODE[str(Feature.MODEL_CUSTOMIZATION_NOVA)]
)
else:
feature_list.append(
FEATURE_TO_CODE[str(Feature.MODEL_CUSTOMIZATION_OSS)]
)
except Exception: # pylint: disable=W0703
logger.debug(
"Unable to determine NOVA/OSS model type for telemetry."
)
if (
hasattr(sagemaker_session, "sagemaker_config")
and sagemaker_session.sagemaker_config
and feature != Feature.SDK_DEFAULTS
):
feature_list.append(FEATURE_TO_CODE[str(Feature.SDK_DEFAULTS)])
if (
hasattr(sagemaker_session, "local_mode")
and sagemaker_session.local_mode
and feature != Feature.LOCAL_MODE
):
feature_list.append(FEATURE_TO_CODE[str(Feature.LOCAL_MODE)])
# Construct the extra info to track platform and environment usage metadata
extra = (
f"{func_name}"
f"&x-sdkVersion={SDK_VERSION}"
f"&x-env={PYTHON_VERSION}"
f"&x-sys={OS_NAME_VERSION}"
f"&x-platform={studio_app_type}"
)
# Add endpoint ARN to the extra info if available
if hasattr(sagemaker_session, "endpoint_arn") and sagemaker_session.endpoint_arn:
extra += f"&x-endpointArn={sagemaker_session.endpoint_arn}"
# Add created_by from environment variable if available
created_by = os.environ.get(_CREATED_BY_ENV_VAR, "")
if created_by:
extra += f"&x-createdBy={quote(created_by, safe='')}"
# Extract granular telemetry params from the instance
if telemetry_params and len(args) > 0:
extra += _extract_telemetry_params(args[0], kwargs, telemetry_params)
start_timer = perf_counter()
try:
# Call the original function
response = func(*args, **kwargs)
stop_timer = perf_counter()
elapsed = stop_timer - start_timer
extra += f"&x-latency={round(elapsed, 2)}"
# For specified response types (e.g., TrainingJob), obtain the ARN of the
# resource created if present so that it can be included.
resource_arn = get_resource_arn(response)
if resource_arn:
extra += f"&x-resourceArn={resource_arn}"
if not telemetry_opt_out_flag:
_send_telemetry_request(
STATUS_TO_CODE[str(Status.SUCCESS)],
feature_list,
sagemaker_session,
None,
None,
extra,
)
except Exception as e: # pylint: disable=W0703
stop_timer = perf_counter()
elapsed = stop_timer - start_timer
extra += f"&x-latency={round(elapsed, 2)}"
extra += f"&x-errorCategory={_classify_error(e)}"
if not telemetry_opt_out_flag:
_send_telemetry_request(
STATUS_TO_CODE[str(Status.FAILURE)],
feature_list,
sagemaker_session,
str(e),
e.__class__.__name__,
extra,
)
caught_ex = e
finally:
if caught_ex:
raise caught_ex
return response # pylint: disable=W0150
else:
logger.debug(
"Unable to send telemetry for function %s. "
"sagemaker_session is not provided or not valid.",
func_name,
)
return func(*args, **kwargs)
return wrapper
return decorator
def _send_telemetry_request(
status: int,
feature_list: List[int],
session: Session,
failure_reason: str = None,
failure_type: str = None,
extra_info: str = None,
) -> None:
"""Make GET request to an empty object in S3 bucket"""
try:
accountId = _get_accountId(session) if session else "NotAvailable"
region = _get_region_or_default(session)
try:
Region(region) # Validate the region
except ValueError:
logger.warning(
"Region not found in supported regions. Telemetry request will not be emitted."
)
return
url = _construct_url(
accountId,
region,
str(status),
str(
",".join(map(str, feature_list))
), # Remove brackets and quotes to cut down on length
failure_reason,
failure_type,
extra_info,
)
# Send the telemetry request
logger.debug("Sending telemetry request to [%s]", url)
_requests_helper(url, 2)
logger.debug("SageMaker Python SDK telemetry successfully emitted.")
except Exception: # pylint: disable=W0703
logger.debug("SageMaker Python SDK telemetry not emitted!")
def _construct_url(
accountId: str,
region: str,
status: str,
feature: str,
failure_reason: str,
failure_type: str,
extra_info: str,
) -> str:
"""Construct the URL for the telemetry request"""
base_url = (
f"https://sm-pysdk-t-{region}.s3.{region}.amazonaws.com/telemetry?"
f"x-accountId={accountId}"
f"&x-status={status}"
f"&x-feature={feature}"
)
logger.debug("Failure reason: %s", failure_reason)
if failure_reason:
base_url += f"&x-failureReason={failure_reason}"
base_url += f"&x-failureType={failure_type}"
if extra_info:
base_url += f"&x-extra={extra_info}"
return base_url
def _requests_helper(url, timeout):
"""Make a GET request to the given URL"""
response = None
try:
response = requests.get(url, timeout)
except requests.exceptions.RequestException as e:
logger.exception("Request exception: %s", str(e))
return response
def _get_accountId(session):
"""Return the account ID from the boto session"""
try:
sts = session.boto_session.client("sts")
return sts.get_caller_identity()["Account"]
except Exception: # pylint: disable=W0703
return None
def _get_region_or_default(session):
"""Return the region name from the boto session or default to us-west-2"""
try:
return session.boto_session.region_name
except Exception: # pylint: disable=W0703
return DEFAULT_AWS_REGION
def _get_default_sagemaker_session():
"""Return the default sagemaker session"""
boto_session = boto3.Session(region_name=DEFAULT_AWS_REGION)
sagemaker_session = Session(boto_session=boto_session)
return sagemaker_session