test: fix broken and flaky CI Health V2 integ tests

tests/integ/sagemaker/jumpstart/model/test_jumpstart_model.py

@@ -167,6 +167,7 @@ def test_jumpstart_gated_model_neuron(setup):
     assert response is not None
 
 
+@x_fail_if_ice
 def test_jumpstart_gated_model(setup):
 
     model_id = "meta-textgeneration-llama-2-7b"

tests/integ/sagemaker/jumpstart/private_hub/estimator/test_jumpstart_private_hub_estimator.py

@@ -30,6 +30,7 @@
     get_sm_session,
     get_training_dataset_for_model_and_version,
     add_model_references_to_hub,
+    x_fail_if_ice,
 )
 
 MAX_INIT_TIME_SECONDS = 5
@@ -135,6 +136,7 @@ def test_jumpstart_hub_estimator_with_session(setup, add_model_references):
 
 
 @pytest.mark.slow_test
+@x_fail_if_ice
 def test_jumpstart_hub_gated_estimator_with_eula(setup, add_model_references):
 
     model_id, model_version = "meta-textgeneration-llama-2-7b", "*"

tests/integ/sagemaker/jumpstart/private_hub/model/test_jumpstart_private_hub_model.py

@@ -31,6 +31,7 @@
 from tests.integ.sagemaker.jumpstart.utils import (
     get_sm_session,
     add_model_references_to_hub,
+    x_fail_if_ice,
 )
 
 MAX_INIT_TIME_SECONDS = 5
@@ -91,6 +92,7 @@ def test_jumpstart_hub_model_with_default_session(setup, add_model_references):
 
 
 @pytest.mark.slow_test
+@x_fail_if_ice
 def test_jumpstart_hub_gated_model(setup, add_model_references):
 
     model_id = "meta-textgeneration-llama-3-2-1b"

tests/integ/sagemaker/lineage/conftest.py

@@ -662,6 +662,12 @@ def static_endpoint_context(sagemaker_session, static_pipeline_execution_arn):
         )
         endpoint_arn = get_endpoint_arn_from_static_pipeline(sagemaker_session)
 
+    if endpoint_arn is None:
+        raise Exception(
+            f"Static endpoint {STATIC_ENDPOINT_NAME} does not exist and could not be deployed "
+            f"from static pipeline execution {static_pipeline_execution_arn}."
+        )
+
     contexts = sagemaker_session.sagemaker_client.list_contexts(SourceUri=endpoint_arn)[
         "ContextSummaries"
     ]
@@ -888,11 +894,36 @@ def _deploy_static_endpoint(execution_arn, sagemaker_session):
         model_package.deploy(1, "ml.m5.xlarge", endpoint_name=STATIC_ENDPOINT_NAME)
         time.sleep(120)
     except ClientError as e:
-        if e.response["Error"]["Code"] == "ValidationException":
-            print(f"Endpoint {STATIC_ENDPOINT_NAME} already exists. Continuing.")
-            pass
-        else:
+        if e.response["Error"]["Code"] != "ValidationException":
             raise (e)
+        if get_endpoint_arn_from_static_pipeline(sagemaker_session) is not None:
+            print(f"Endpoint {STATIC_ENDPOINT_NAME} already exists. Continuing.")
+            return
+        # The endpoint is gone but an endpoint config with the same name outlived it (e.g. the
+        # endpoint was deleted out of band), so deploy() failed on CreateEndpointConfig without
+        # creating an endpoint. Recreate the endpoint from the existing endpoint config.
+        print(
+            f"Endpoint {STATIC_ENDPOINT_NAME} is missing but its endpoint config exists. "
+            "Recreating the endpoint from the existing endpoint config."
+        )
+        _create_static_endpoint_from_existing_config(sagemaker_session)
+
+
+def _create_static_endpoint_from_existing_config(sagemaker_session):
+    try:
+        sagemaker_session.sagemaker_client.create_endpoint(
+            EndpointName=STATIC_ENDPOINT_NAME, EndpointConfigName=STATIC_ENDPOINT_NAME
+        )
+    except ClientError as e:
+        # Another test worker may have recreated the endpoint concurrently; anything else
+        # (e.g. the endpoint config is missing too) is a real failure.
+        if (
+            e.response["Error"]["Code"] != "ValidationException"
+            or get_endpoint_arn_from_static_pipeline(sagemaker_session) is None
+        ):
+            raise
+    sagemaker_session.wait_for_endpoint(STATIC_ENDPOINT_NAME)
+    time.sleep(120)
 
 
 @pytest.fixture

tests/integ/sagemaker/serve/test_base_model_builder_deploy.py

@@ -14,7 +14,6 @@
 
 import os
 import uuid
-from typing import Generator
 
 import numpy as np
 import pandas as pd
@@ -39,43 +38,43 @@
 from sagemaker.serve import InferenceSpec, SchemaBuilder
 from sagemaker.serve.builder.model_builder import ModelBuilder
 from sagemaker.serverless.serverless_inference_config import ServerlessInferenceConfig
-from tests.integ.utils import cleanup_model_resources
 
 
-@pytest.fixture(autouse=True)
-def cleanup_endpoints(mb_sagemaker_session) -> Generator[None, None, None]:
-    """Clean up any existing endpoints before and after tests."""
-    sagemaker_client = mb_sagemaker_session.sagemaker_client
+def _cleanup_deployment(model_builder, endpoint_name):
+    """Best-effort delete of the endpoint, endpoint config and model one test created.
 
-    # Pre-test cleanup
+    Only touch resources created under ``endpoint_name``. The integ suites run concurrently
+    in a shared account, so sweeping every endpoint there deletes endpoints that other running
+    tests and long-lived static fixtures (e.g. the lineage tests' static endpoint) depend on.
+    """
+    sagemaker_client = model_builder.sagemaker_session.sagemaker_client
+    endpoint_config_name = endpoint_name
     try:
-        endpoints = sagemaker_client.list_endpoints()
-        for endpoint in endpoints["Endpoints"]:
-            try:
-                sagemaker_client.delete_endpoint(EndpointName=endpoint["EndpointName"])
-                sagemaker_client.delete_endpoint_config(
-                    EndpointConfigName=endpoint["EndpointConfigName"]
-                )
-            except Exception as e:
-                print(f"Error cleaning up endpoint {endpoint['EndpointName']}: {e}")
-    except Exception as e:
-        print(f"Error listing endpoints: {e}")
-
-    yield
-
-    # Post-test cleanup
-    try:
-        endpoints = sagemaker_client.list_endpoints()
-        for endpoint in endpoints["Endpoints"]:
-            try:
-                sagemaker_client.delete_endpoint(EndpointName=endpoint["EndpointName"])
-                sagemaker_client.delete_endpoint_config(
-                    EndpointConfigName=endpoint["EndpointConfigName"]
-                )
-            except Exception as e:
-                print(f"Error cleaning up endpoint {endpoint['EndpointName']}: {e}")
-    except Exception as e:
-        print(f"Error listing endpoints: {e}")
+        endpoint_config_name = sagemaker_client.describe_endpoint(EndpointName=endpoint_name)[
+            "EndpointConfigName"
+        ]
+    except Exception:  # noqa: BLE001 -- the endpoint may never have been created
+        pass
+
+    deletions = [
+        ("endpoint", lambda: sagemaker_client.delete_endpoint(EndpointName=endpoint_name)),
+        (
+            "endpoint config",
+            lambda: sagemaker_client.delete_endpoint_config(
+                EndpointConfigName=endpoint_config_name
+            ),
+        ),
+        (
+            "model",
+            lambda: sagemaker_client.delete_model(ModelName=model_builder.built_model.name),
+        ),
+    ]
+    # Each deletion runs independently, so one failure does not leave the rest behind.
+    for resource, delete in deletions:
+        try:
+            delete()
+        except Exception as e:  # noqa: BLE001 -- best-effort cleanup
+            print(f"Error cleaning up {resource} for endpoint {endpoint_name}: {e}")
 
 
 @pytest.fixture(scope="module")
@@ -184,44 +183,41 @@ def invoke(self, input_object: object, model: object):
 
 
 def test_real_time_deployment(xgboost_model_builder):
-    real_time_predictor = xgboost_model_builder.deploy(
-        endpoint_name=f"test-{uuid.uuid1().hex}", initial_instance_count=1
-    )
+    endpoint_name = f"test-{uuid.uuid1().hex}"
+    try:
+        real_time_predictor = xgboost_model_builder.deploy(
+            endpoint_name=endpoint_name, initial_instance_count=1
+        )
 
-    assert real_time_predictor is not None
-    cleanup_model_resources(
-        sagemaker_session=xgboost_model_builder.sagemaker_session,
-        model_name=xgboost_model_builder.built_model.name,
-        endpoint_name=xgboost_model_builder.built_model.endpoint_name,
-    )
+        assert real_time_predictor is not None
+    finally:
+        _cleanup_deployment(xgboost_model_builder, endpoint_name)
 
 
 def test_serverless_deployment(xgboost_model_builder):
-    serverless_predictor = xgboost_model_builder.deploy(
-        endpoint_name=f"test1-{uuid.uuid1().hex}", inference_config=ServerlessInferenceConfig()
-    )
+    endpoint_name = f"test1-{uuid.uuid1().hex}"
+    try:
+        serverless_predictor = xgboost_model_builder.deploy(
+            endpoint_name=endpoint_name, inference_config=ServerlessInferenceConfig()
+        )
 
-    assert serverless_predictor is not None
-    cleanup_model_resources(
-        sagemaker_session=xgboost_model_builder.sagemaker_session,
-        model_name=xgboost_model_builder.built_model.name,
-        endpoint_name=xgboost_model_builder.built_model.endpoint_name,
-    )
+        assert serverless_predictor is not None
+    finally:
+        _cleanup_deployment(xgboost_model_builder, endpoint_name)
 
 
 def test_async_deployment(xgboost_model_builder, mb_sagemaker_session):
-    async_predictor = xgboost_model_builder.deploy(
-        endpoint_name=f"test2-{uuid.uuid1().hex}",
-        inference_config=AsyncInferenceConfig(
-            output_path=s3_path_join(
-                "s3://", mb_sagemaker_session.default_bucket(), "async_inference/output"
-            )
-        ),
-    )
+    endpoint_name = f"test2-{uuid.uuid1().hex}"
+    try:
+        async_predictor = xgboost_model_builder.deploy(
+            endpoint_name=endpoint_name,
+            inference_config=AsyncInferenceConfig(
+                output_path=s3_path_join(
+                    "s3://", mb_sagemaker_session.default_bucket(), "async_inference/output"
+                )
+            ),
+        )
 
-    assert async_predictor is not None
-    cleanup_model_resources(
-        sagemaker_session=xgboost_model_builder.sagemaker_session,
-        model_name=xgboost_model_builder.built_model.name,
-        endpoint_name=xgboost_model_builder.built_model.endpoint_name,
-    )
+        assert async_predictor is not None
+    finally:
+        _cleanup_deployment(xgboost_model_builder, endpoint_name)

tests/integ/sagemaker/workflow/test_deployment_steps.py

@@ -27,6 +27,7 @@
 import os
 import time
 import uuid
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -39,6 +40,10 @@
 INSTANCE_TYPE = "ml.m5.xlarge"
 EXECUTION_TIMEOUT_SECONDS = 45 * 60
 POLL_SECONDS = 30
+# Largest page size the List* APIs accept.
+LIST_PAGE_SIZE = 100
+# Margin on the CreationTimeAfter filter to absorb clock skew with the service.
+CREATION_TIME_MARGIN = timedelta(minutes=10)
 
 
 # sagemaker_session and pipeline_session come from tests/conftest.py, and role from
@@ -50,6 +55,7 @@
 def test_deployment_steps_execute_end_to_end(sagemaker_session, pipeline_session, role):
     """Chained EndpointConfig -> Endpoint -> InferenceComponent pipeline run."""
     stamp = uuid.uuid4().hex[:8]
+    created_after = datetime.now(timezone.utc) - CREATION_TIME_MARGIN
     config_name = f"integ-deploy-cfg-{stamp}"
     endpoint_name = f"integ-deploy-ep-{stamp}"
     component_name = f"integ-deploy-ic-{stamp}"
@@ -155,14 +161,21 @@ def test_deployment_steps_execute_end_to_end(sagemaker_session, pipeline_session
 
         # Resolve the actual server-side names (the steps suffix them).
         actual_config = _resolve(
-            sm_client.list_endpoint_configs(NameContains=stamp)["EndpointConfigs"],
+            _list_all(sm_client, "list_endpoint_configs", "EndpointConfigs", stamp, created_after),
             "EndpointConfigName",
         )
         actual_endpoint = _resolve(
-            sm_client.list_endpoints(NameContains=stamp)["Endpoints"], "EndpointName"
+            _list_all(sm_client, "list_endpoints", "Endpoints", stamp, created_after),
+            "EndpointName",
         )
         actual_component = _resolve(
-            sm_client.list_inference_components(NameContains=stamp)["InferenceComponents"],
+            _list_all(
+                sm_client,
+                "list_inference_components",
+                "InferenceComponents",
+                stamp,
+                created_after,
+            ),
             "InferenceComponentName",
         )
 
@@ -180,7 +193,26 @@ def test_deployment_steps_execute_end_to_end(sagemaker_session, pipeline_session
         )
         assert component_desc["EndpointName"] == actual_endpoint
     finally:
-        _cleanup(sagemaker_session, sm_client, stamp, pipeline, model_name)
+        _cleanup(sagemaker_session, sm_client, stamp, created_after, pipeline, model_name)
+
+
+def _list_all(sm_client, operation, result_key, stamp, created_after):
+    """Return every resource whose name contains ``stamp``, across all result pages.
+
+    SageMaker applies ``NameContains`` to each page of results rather than before paging,
+    so a single call can return an empty page (plus a NextToken) even though a matching
+    resource exists. That happens whenever more than a page of other resources were created
+    after this run's, which is routine while the rest of the integ suite runs in parallel in
+    the same account, so always walk every page. ``CreationTimeAfter`` and the larger page
+    size keep the number of pages small.
+    """
+    paginator = sm_client.get_paginator(operation)
+    pages = paginator.paginate(
+        NameContains=stamp,
+        CreationTimeAfter=created_after,
+        PaginationConfig={"PageSize": LIST_PAGE_SIZE},
+    )
+    return [item for page in pages for item in page[result_key]]
 
 
 def _resolve(items, key):
@@ -189,21 +221,25 @@ def _resolve(items, key):
     return items[0][key]
 
 
-def _cleanup(sagemaker_session, sm_client, stamp, pipeline, model_name):
+def _cleanup(sagemaker_session, sm_client, stamp, created_after, pipeline, model_name):
     """Delete every resource this run created, in dependency order."""
-    for component in sm_client.list_inference_components(NameContains=stamp)["InferenceComponents"]:
+    for component in _list_all(
+        sm_client, "list_inference_components", "InferenceComponents", stamp, created_after
+    ):
         name = component["InferenceComponentName"]
         try:
             sm_client.delete_inference_component(InferenceComponentName=name)
             _wait_component_deleted(sm_client, name)
         except Exception:  # noqa: BLE001 -- best-effort cleanup
             pass
-    for endpoint in sm_client.list_endpoints(NameContains=stamp)["Endpoints"]:
+    for endpoint in _list_all(sm_client, "list_endpoints", "Endpoints", stamp, created_after):
         try:
             sm_client.delete_endpoint(EndpointName=endpoint["EndpointName"])
         except Exception:  # noqa: BLE001
             pass
-    for config in sm_client.list_endpoint_configs(NameContains=stamp)["EndpointConfigs"]:
+    for config in _list_all(
+        sm_client, "list_endpoint_configs", "EndpointConfigs", stamp, created_after
+    ):
         try:
             sm_client.delete_endpoint_config(EndpointConfigName=config["EndpointConfigName"])
         except Exception:  # noqa: BLE001

tests/integ/test_huggingface_torch_distributed.py

@@ -16,9 +16,11 @@
 import pytest
 from sagemaker.huggingface import HuggingFace
 from tests.integ import DATA_DIR, TRAINING_DEFAULT_TIMEOUT_MINUTES, timeout
+from tests.integ.sagemaker.jumpstart.utils import fit_estimator_with_capacity_xfail, x_fail_if_ice
 
 
 @pytest.mark.slow_test
+@x_fail_if_ice
 def test_huggingface_torch_distributed_g5_glue(
     sagemaker_session,
     huggingface_training_latest_version,
@@ -49,4 +51,7 @@ def test_huggingface_torch_distributed_g5_glue(
             sagemaker_session=sagemaker_session,
             disable_profiler=True,
         )
-        estimator.fit()
+        # ml.g5.4xlarge capacity is often unavailable, and a job waiting for capacity never
+        # fails on its own: it just sits in "Pending" until the timeout above kills the test.
+        # Fail fast with a CapacityError instead, which x_fail_if_ice turns into an xfail.
+        fit_estimator_with_capacity_xfail(estimator, inputs=None)