preparing test

Author: sanderegg

packages/pytest-simcore/src/pytest_simcore/helpers/aws_ec2.py

@@ -5,7 +5,12 @@
 from models_library.docker import DockerGenericTag
 from types_aiobotocore_ec2 import EC2Client
 from types_aiobotocore_ec2.literals import InstanceStateNameType, InstanceTypeType
-from types_aiobotocore_ec2.type_defs import FilterTypeDef, InstanceTypeDef, TagTypeDef
+from types_aiobotocore_ec2.type_defs import (
+    FilterTypeDef,
+    InstanceTypeDef,
+    ReservationTypeDef,
+    TagTypeDef,
+)
 
 
 async def assert_autoscaled_computational_ec2_instances(
@@ -72,6 +77,7 @@ async def assert_autoscaled_dynamic_warm_pools_ec2_instances(
     expected_additional_tag_keys: list[str],
     expected_pre_pulled_images: list[DockerGenericTag] | None,
     instance_filters: Sequence[FilterTypeDef] | None,
+    check_reservation_index: int | None = None,
 ) -> list[InstanceTypeDef]:
     return await assert_ec2_instances(
         ec2_client,
@@ -88,9 +94,80 @@ async def assert_autoscaled_dynamic_warm_pools_ec2_instances(
         expected_pre_pulled_images=expected_pre_pulled_images,
         expected_user_data=[],
         instance_filters=instance_filters,
+        check_reservation_index=check_reservation_index,
     )
 
 
+async def _assert_reservation(
+    ec2_client: EC2Client,
+    reservation: ReservationTypeDef,
+    *,
+    expected_num_instances: int,
+    expected_instance_type: InstanceTypeType,
+    expected_instance_state: InstanceStateNameType,
+    expected_instance_tag_keys: list[str],
+    expected_user_data: list[str],
+    expected_pre_pulled_images: list[DockerGenericTag] | None,
+) -> list[InstanceTypeDef]:
+    list_instances: list[InstanceTypeDef] = []
+    assert "Instances" in reservation
+    assert (
+        len(reservation["Instances"]) == expected_num_instances
+    ), f"expected {expected_num_instances}, found {len(reservation['Instances'])}"
+    for instance in reservation["Instances"]:
+        assert "InstanceType" in instance
+        assert instance["InstanceType"] == expected_instance_type
+        assert "Tags" in instance
+        assert instance["Tags"]
+        expected_tag_keys = {
+            *expected_instance_tag_keys,
+            "io.simcore.autoscaling.version",
+            "Name",
+        }
+        instance_tag_keys = {tag["Key"] for tag in instance["Tags"] if "Key" in tag}
+        assert instance_tag_keys == expected_tag_keys
+
+        if expected_pre_pulled_images is None:
+            assert "io.simcore.autoscaling.pre_pulled_images" not in instance_tag_keys
+        else:
+            assert "io.simcore.autoscaling.pre_pulled_images" in instance_tag_keys
+
+            def _by_pre_pull_image(ec2_tag: TagTypeDef) -> bool:
+                assert "Key" in ec2_tag
+                return ec2_tag["Key"] == "io.simcore.autoscaling.pre_pulled_images"
+
+            instance_pre_pulled_images_aws_tag = next(
+                iter(filter(_by_pre_pull_image, instance["Tags"]))
+            )
+            assert "Value" in instance_pre_pulled_images_aws_tag
+            assert (
+                instance_pre_pulled_images_aws_tag["Value"]
+                == f"{json_dumps(expected_pre_pulled_images)}"
+            )
+
+        assert "PrivateDnsName" in instance
+        instance_private_dns_name = instance["PrivateDnsName"]
+        if expected_instance_state not in ["terminated"]:
+            # NOTE: moto behaves here differently than AWS by still returning an IP which does not really make sense
+            assert instance_private_dns_name.endswith(".ec2.internal")
+        assert "State" in instance
+        state = instance["State"]
+        assert "Name" in state
+        assert state["Name"] == expected_instance_state
+
+        assert "InstanceId" in instance
+        user_data = await ec2_client.describe_instance_attribute(
+            Attribute="userData", InstanceId=instance["InstanceId"]
+        )
+        assert "UserData" in user_data
+        assert "Value" in user_data["UserData"]
+        user_data = base64.b64decode(user_data["UserData"]["Value"]).decode()
+        for user_data_string in expected_user_data:
+            assert user_data.count(user_data_string) == 1
+        list_instances.append(instance)
+    return list_instances
+
+
 async def assert_ec2_instances(
     ec2_client: EC2Client,
     *,
@@ -102,66 +179,35 @@ async def assert_ec2_instances(
     expected_user_data: list[str],
     expected_pre_pulled_images: list[DockerGenericTag] | None = None,
     instance_filters: Sequence[FilterTypeDef] | None = None,
+    check_reservation_index: int | None = None,
 ) -> list[InstanceTypeDef]:
-    list_instances: list[InstanceTypeDef] = []
     all_instances = await ec2_client.describe_instances(Filters=instance_filters or [])
     assert len(all_instances["Reservations"]) == expected_num_reservations
+    if check_reservation_index is not None:
+        assert check_reservation_index < len(all_instances["Reservations"])
+        reservation = all_instances["Reservations"][check_reservation_index]
+        return await _assert_reservation(
+            ec2_client,
+            reservation,
+            expected_num_instances=expected_num_instances,
+            expected_instance_type=expected_instance_type,
+            expected_instance_state=expected_instance_state,
+            expected_instance_tag_keys=expected_instance_tag_keys,
+            expected_user_data=expected_user_data,
+            expected_pre_pulled_images=expected_pre_pulled_images,
+        )
+    list_instances: list[InstanceTypeDef] = []
     for reservation in all_instances["Reservations"]:
-        assert "Instances" in reservation
-        assert (
-            len(reservation["Instances"]) == expected_num_instances
-        ), f"expected {expected_num_instances}, found {len(reservation['Instances'])}"
-        for instance in reservation["Instances"]:
-            assert "InstanceType" in instance
-            assert instance["InstanceType"] == expected_instance_type
-            assert "Tags" in instance
-            assert instance["Tags"]
-            expected_tag_keys = {
-                *expected_instance_tag_keys,
-                "io.simcore.autoscaling.version",
-                "Name",
-            }
-            instance_tag_keys = {tag["Key"] for tag in instance["Tags"] if "Key" in tag}
-            assert instance_tag_keys == expected_tag_keys
-
-            if expected_pre_pulled_images is None:
-                assert (
-                    "io.simcore.autoscaling.pre_pulled_images" not in instance_tag_keys
-                )
-            else:
-                assert "io.simcore.autoscaling.pre_pulled_images" in instance_tag_keys
-
-                def _by_pre_pull_image(ec2_tag: TagTypeDef) -> bool:
-                    assert "Key" in ec2_tag
-                    return ec2_tag["Key"] == "io.simcore.autoscaling.pre_pulled_images"
-
-                instance_pre_pulled_images_aws_tag = next(
-                    iter(filter(_by_pre_pull_image, instance["Tags"]))
-                )
-                assert "Value" in instance_pre_pulled_images_aws_tag
-                assert (
-                    instance_pre_pulled_images_aws_tag["Value"]
-                    == f"{json_dumps(expected_pre_pulled_images)}"
-                )
-
-            assert "PrivateDnsName" in instance
-            instance_private_dns_name = instance["PrivateDnsName"]
-            if expected_instance_state not in ["terminated"]:
-                # NOTE: moto behaves here differently than AWS by still returning an IP which does not really make sense
-                assert instance_private_dns_name.endswith(".ec2.internal")
-            assert "State" in instance
-            state = instance["State"]
-            assert "Name" in state
-            assert state["Name"] == expected_instance_state
-
-            assert "InstanceId" in instance
-            user_data = await ec2_client.describe_instance_attribute(
-                Attribute="userData", InstanceId=instance["InstanceId"]
+        list_instances.extend(
+            await _assert_reservation(
+                ec2_client,
+                reservation,
+                expected_num_instances=expected_num_instances,
+                expected_instance_type=expected_instance_type,
+                expected_instance_state=expected_instance_state,
+                expected_instance_tag_keys=expected_instance_tag_keys,
+                expected_user_data=expected_user_data,
+                expected_pre_pulled_images=expected_pre_pulled_images,
             )
-            assert "UserData" in user_data
-            assert "Value" in user_data["UserData"]
-            user_data = base64.b64decode(user_data["UserData"]["Value"]).decode()
-            for user_data_string in expected_user_data:
-                assert user_data.count(user_data_string) == 1
-            list_instances.append(instance)
+        )
     return list_instances

services/autoscaling/tests/unit/conftest.py

@@ -1093,7 +1093,7 @@ def ec2_instances_allowed_types_with_only_1_buffered(
             "t2.micro": EC2InstanceBootSpecific(
                 ami_id=faker.pystr(),
                 pre_pull_images=fake_pre_pull_images,
-                buffer_count=faker.pyint(min_value=1, max_value=10),
+                buffer_count=faker.pyint(min_value=2, max_value=10),
             )
         }
 

services/autoscaling/tests/unit/test_modules_auto_scaling_dynamic.py

@@ -2061,3 +2061,111 @@ async def test_warm_buffers_are_started_to_replace_missing_hot_buffers(
         len(analyzed_cluster.pending_ec2s)
         == app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MACHINES_BUFFER
     )
+
+
+@pytest.mark.parametrize(
+    # NOTE: only the main test test_cluster_scaling_up_and_down is run with all options
+    "with_docker_join_drained",
+    ["without_AUTOSCALING_DOCKER_JOIN_DRAINED"],
+    indirect=True,
+)
+@pytest.mark.parametrize(
+    # NOTE: only the main test test_cluster_scaling_up_and_down is run with all options
+    "with_drain_nodes_labelled",
+    ["with_AUTOSCALING_DRAIN_NODES_WITH_LABELS"],
+    indirect=True,
+)
+async def test_warm_buffers_only_replace_hot_buffer_if_service_is_started_issue7071(
+    patch_ec2_client_launch_instances_min_number_of_instances: mock.Mock,
+    minimal_configuration: None,
+    with_instances_machines_hot_buffer: EnvVarsDict,
+    ec2_client: EC2Client,
+    initialized_app: FastAPI,
+    app_settings: ApplicationSettings,
+    ec2_instance_custom_tags: dict[str, str],
+    buffer_count: int,
+    create_buffer_machines: Callable[
+        [int, InstanceTypeType, InstanceStateNameType, list[DockerGenericTag] | None],
+        Awaitable[list[str]],
+    ],
+    spied_cluster_analysis: MockType,
+    instance_type_filters: Sequence[FilterTypeDef],
+    mock_find_node_with_name_returns_fake_node: mock.Mock,
+    mock_compute_node_used_resources: mock.Mock,
+    mock_docker_tag_node: mock.Mock,
+):
+    # NOTE: https://github.com/ITISFoundation/osparc-simcore/issues/7071
+
+    # pre-requisites
+    assert app_settings.AUTOSCALING_EC2_INSTANCES
+    assert app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MACHINES_BUFFER > 0
+
+    # we have nothing running now
+    all_instances = await ec2_client.describe_instances()
+    assert not all_instances["Reservations"]
+
+    # ensure we get our running hot buffer
+    await auto_scale_cluster(
+        app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
+    )
+    await assert_autoscaled_dynamic_ec2_instances(
+        ec2_client,
+        expected_num_reservations=1,
+        expected_num_instances=app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MACHINES_BUFFER,
+        expected_instance_type=cast(
+            InstanceTypeType,
+            next(
+                iter(app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES)
+            ),
+        ),
+        expected_instance_state="running",
+        expected_additional_tag_keys=list(ec2_instance_custom_tags),
+        instance_filters=instance_type_filters,
+    )
+    # calling again should attach the new nodes to the reserve, but nothing should start
+    await auto_scale_cluster(
+        app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
+    )
+    await assert_autoscaled_dynamic_ec2_instances(
+        ec2_client,
+        expected_num_reservations=1,
+        expected_num_instances=app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MACHINES_BUFFER,
+        expected_instance_type=cast(
+            InstanceTypeType,
+            next(
+                iter(app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES)
+            ),
+        ),
+        expected_instance_state="running",
+        expected_additional_tag_keys=list(ec2_instance_custom_tags),
+        instance_filters=instance_type_filters,
+    )
+
+    # have a few warm buffers ready with the same type as the hot buffer machines
+    buffer_machines = await create_buffer_machines(
+        buffer_count,
+        cast(
+            InstanceTypeType,
+            next(
+                iter(app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES)
+            ),
+        ),
+        "stopped",
+        None,
+    )
+    await assert_autoscaled_dynamic_warm_pools_ec2_instances(
+        ec2_client,
+        expected_num_reservations=2,
+        check_reservation_index=1,
+        expected_num_instances=buffer_count,
+        expected_instance_type=cast(
+            InstanceTypeType,
+            next(
+                iter(app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES)
+            ),
+        ),
+        expected_instance_state="stopped",
+        expected_additional_tag_keys=list(ec2_instance_custom_tags),
+        expected_pre_pulled_images=None,
+        instance_filters=None,
+    )