# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See LICENSE for more details.
#
# Copyright (c) 2017 ScyllaDB

from __future__ import absolute_import, annotations
import logging
import os
import time
import ipaddress
import tempfile
from collections import defaultdict
from contextlib import contextmanager

from botocore.exceptions import ClientError
import boto3
import google.api_core.exceptions
import oci
from google.cloud.compute_v1.types import Instance as GceInstance
from mypy_boto3_ec2 import EC2Client

from sdcm.cloud_api_client import ScyllaCloudAPIClient, ScyllaCloudAPIError
from sdcm.provision.oci.constants import TAG_NAMESPACE
from sdcm.provision.aws.capacity_reservation import SCTCapacityReservation
from sdcm.provision.aws.dedicated_host import SCTDedicatedHosts
from sdcm.provision.aws.emr_provisioner import list_emr_clusters
from sdcm.provision.azure.provisioner import AzureProvisioner
from sdcm.utils.argus import (
    ArgusError,
    ReplayOnlyArgusSCTClient,
    get_argus_client,
    get_argus_use_tunnel_from_env,
    terminate_resource_in_argus,
)
from sdcm.utils.aws_kms import AwsKms
from sdcm.utils.aws_region import AwsRegion
from sdcm.utils.gce_region import GceRegion
from sdcm.sct_config import AWS_SUPPORTED_REGIONS
from sdcm.utils.common import (
    all_aws_regions,
    aws_tags_to_dict,
    get_post_behavior_actions,
    get_testrun_status,
    list_cloudformation_stacks_aws,
    list_clusters_gke,
    list_clusters_eks,
    list_elastic_ips_aws,
    list_instances_aws,
    list_instances_gce,
    list_launch_templates_aws,
    list_load_balancers_aws,
    list_placement_groups_aws,
    list_resources_docker,
    list_test_security_groups,
    run_per_region_ignore_failures,
)
from sdcm.utils.parallel_object import ParallelObject
from sdcm.utils.context_managers import environment
from sdcm.utils.decorators import retrying
from sdcm.utils.gce_utils import (
    GkeCleaner,
    get_gce_compute_instances_client,
)
from sdcm.utils.oci_utils import (
    delete_oci_volume_with_retry,
    OciService,
    oci_keep_action,
    SUPPORTED_REGIONS,
    get_oci_compartment_id,
    list_instances_oci,
)


LOGGER = logging.getLogger("utils")


def init_argus_client(test_id: str, use_tunnel: bool | None = None, log_dir: str | None = None):
    if use_tunnel is None:
        use_tunnel = get_argus_use_tunnel_from_env()
    try:
        argus_client = get_argus_client(run_id=test_id, use_tunnel=use_tunnel, log_dir=log_dir)
    except ArgusError as exc:
        LOGGER.warning("Unable to initialize Argus: %s", exc.message)
        # Fall back to replay-only mode so all intended API calls are still
        # captured to the replay log instead of being silently mocked away.
        argus_client = ReplayOnlyArgusSCTClient(
            run_id=test_id,
            log_dir=log_dir or tempfile.gettempdir(),
        )
    return argus_client


def clean_cloud_resources(tags_dict, config=None, dry_run=False):
    """
    Clean up cloud resources created in various cloud platforms.

    :param config: instance of the 'SCTConfiguration' class
    :param tags_dict: key-value pairs used for filtering
    :param dry_run: boolean value which defines whether we should really cleanup resources or not
    :return: True when the cleanup flow completed, False when neither TestId nor RunByUser tag is
        provided, None for local K8S backends (no remote resources to clean)
    """
    if "TestId" not in tags_dict and "RunByUser" not in tags_dict:
        LOGGER.error("Can't clean cloud resources, TestId or RunByUser is missing")
        return False

    failures: list[tuple[str, Exception]] = []

    @contextmanager
    def cleanup_step(name):
        """Run a cleanup step in isolation so a failure does not stop the rest of the cleanup."""
        try:
            yield
        except Exception as exc:  # noqa: BLE001
            LOGGER.warning("Cleanup step '%s' failed, skipping: %s", name, exc)
            failures.append((name, exc))

    cluster_backend = config.get("cluster_backend") or ""
    aws_regions = config.region_names
    if config.get("fallback_to_next_region"):
        # region fallback may have relocated the cluster, so scan the fallback-eligible regions, not
        # every AWS region (which would hit unreachable opt-in regions e.g. me-south-1 and time out).
        aws_regions = sorted(set(AWS_SUPPORTED_REGIONS) | set(config.region_names or []))
        LOGGER.info(
            "fallback_to_next_region enabled: scanning fallback-eligible AWS regions for cleanup: %s", aws_regions
        )
    gce_projects = [config.get("gce_project") or "gcp-sct-project-1"]

    if cluster_backend.startswith("k8s-local"):
        LOGGER.info("No remote resources are expected in the local K8S setups. Skipping.")
        return
    if cluster_backend in ("k8s-eks", ""):
        for name, clean_func in (
            ("EKS clusters", clean_clusters_eks),
            ("AWS launch templates", clean_launch_templates_aws),
            ("AWS load balancers", clean_load_balancers_aws),
            ("AWS CloudFormation stacks", clean_cloudformation_stacks_aws),
        ):
            with cleanup_step(name):
                clean_func(tags_dict, regions=aws_regions, dry_run=dry_run)
    if cluster_backend in ("k8s-gke", ""):
        for project in gce_projects:
            with environment(SCT_GCE_PROJECT=project):
                with cleanup_step("GKE clusters"):
                    clean_clusters_gke(tags_dict, dry_run=dry_run)
                with cleanup_step("GKE orphaned disks"):
                    clean_orphaned_gke_disks(tags_dict, dry_run=dry_run)

    if cluster_backend in ("aws", ""):
        with cleanup_step("EMR clusters"):
            clean_emr_clusters(tags_dict, regions=aws_regions, dry_run=dry_run)
    if cluster_backend in ("aws", "k8s-eks", ""):
        with cleanup_step("AWS instances"):
            clean_instances_aws(tags_dict, regions=aws_regions, dry_run=dry_run)
        with cleanup_step("AWS capacity reservations"):
            if config.region_names:
                SCTCapacityReservation.get_cr_from_aws(config, force_fetch=True)
                SCTCapacityReservation.cancel(config)
            else:
                SCTCapacityReservation.cancel_all_regions(config.get("test_id"))
            # cleanup capacity reservations across all regions as region fallback can leave an idle CR in an abandoned region
            if not dry_run and (test_id := config.get("test_id")):
                SCTCapacityReservation.cancel_all_regions(test_id)
        with cleanup_step("AWS dedicated hosts"):
            SCTDedicatedHosts.release_by_tags(tags_dict, regions=aws_regions, dry_run=dry_run)
        for name, clean_func in (
            ("AWS elastic IPs", clean_elastic_ips_aws),
            ("AWS security groups", clean_test_security_groups),
            ("AWS placement groups", clean_placement_groups_aws),
        ):
            with cleanup_step(name):
                clean_func(tags_dict, regions=aws_regions, dry_run=dry_run)
        if cluster_backend == "aws" and not dry_run:
            with cleanup_step("AWS KMS aliases"):
                clean_aws_kms_alias(tags_dict, aws_regions or all_aws_regions())
        if not dry_run and (test_id := tags_dict.get("TestId")):
            from sdcm.test_config import TestConfig  # noqa: PLC0415  # avoid import cycle at module load

            with cleanup_step("AWS resolved placement"):
                TestConfig.delete_resolved_placement(test_id)
    if cluster_backend in ("gce", "k8s-gke", ""):
        for project in gce_projects:
            with environment(SCT_GCE_PROJECT=project):
                with cleanup_step("GCE instances"):
                    clean_instances_gce(tags_dict, dry_run=dry_run)
    if cluster_backend in ("azure", ""):
        azure_regions = config.get("azure_region_name") or []
        with cleanup_step("Azure instances"):
            clean_instances_azure(tags_dict, regions=azure_regions, dry_run=dry_run)
    # Always clean local Docker resources for all backends (except k8s-local which has no resources)
    # Tests on any backend may create local Docker containers for stress tools, monitoring, etc.
    if not cluster_backend.startswith("k8s-local"):
        with cleanup_step("Docker resources"):
            clean_resources_docker(tags_dict, dry_run=dry_run)
    if cluster_backend in ("xcloud",):
        with cleanup_step("xcloud clusters"):
            clean_clusters_scylla_cloud(tags_dict, config, dry_run=dry_run)
    if cluster_backend in ("oci", ""):
        with cleanup_step("OCI instances"):
            clean_instances_oci(tags_dict, dry_run=dry_run)
        with cleanup_step("OCI block volumes"):
            clean_orphan_block_volumes_oci(tags_dict=tags_dict, dry_run=dry_run)

    if failures:
        LOGGER.warning(
            "clean_cloud_resources finished with %d skipped step(s): %s",
            len(failures),
            ", ".join(name for name, _ in failures),
        )
    return True


def clean_resources_docker(tags_dict: dict, dry_run: bool = False) -> None:
    """Clean Docker containers and images matching the provided tags.

    Scans the local Docker daemon for containers/images matching tags_dict.

    Args:
        tags_dict: Dictionary of tags to filter resources (must include TestId or RunByUser)
        dry_run: If True, only log what would be deleted without actually deleting
    """
    assert tags_dict, "tags_dict not provided (can't clean all instances)"

    def delete_container(container):
        container.reload()
        container_name = container.name
        LOGGER.info("Cleaned Docker container: %s on `%s'", container_name, container.client.info()["Name"])
        if not dry_run:
            container.remove(v=True, force=True)
            LOGGER.debug("Done.")

    def delete_image(image):
        LOGGER.info("Cleaned Docker image: %s on `%s'", image.tags, image.client.info()["Name"])
        if not dry_run:
            image.client.images.remove(image=image.id, force=True)
            LOGGER.debug("Done.")

    resources_to_clean = list_resources_docker(tags_dict=tags_dict)
    containers = resources_to_clean.get("containers", [])
    images = resources_to_clean.get("images", [])

    if not containers and not images:
        LOGGER.info("There are no resources to clean in Docker")
        return

    for container in containers:
        try:
            delete_container(container)
        except Exception:  # noqa: BLE001
            LOGGER.error("Failed to delete container %s on host `%s'", container, container.client.info()["Name"])

    for image in images:
        try:
            delete_image(image)
        except Exception:  # noqa: BLE001
            LOGGER.error("Failed to delete image tag(s) %s on host `%s'", image.tags, image.client.info()["Name"])


def _oci_tags_match(resource_tags: dict, tags_dict: dict) -> bool:
    for tag_k, tag_v in tags_dict.items():
        if tag_k not in resource_tags:
            return False
        if isinstance(tag_v, list):
            if resource_tags[tag_k] not in tag_v:
                return False
        elif resource_tags[tag_k] != tag_v:
            return False
    return True


def clean_instances_oci(tags_dict: dict, dry_run=False):
    """Remove all instances with specific tags in OCI."""
    if not tags_dict:
        LOGGER.error("tags_dict not provided (can't clean all instances)")
        return

    all_instances = list_instances_oci(tags_dict=tags_dict, verbose=False)
    if not all_instances:
        LOGGER.info("There are no instances to remove in OCI")
        return

    instances_by_region = defaultdict(list)
    for instance in all_instances:
        instances_by_region[instance.region].append(instance)
    for region, instances in instances_by_region.items():
        compute_client = OciService().get_compute_client(region=region)
        for instance in instances:
            tags = (instance.defined_tags or {}).get(TAG_NAMESPACE, {})
            node_type = tags.get("NodeType", "")
            keep_action = oci_keep_action(tags)
            display_name = instance.display_name
            instance_id = instance.id
            test_id = tags.get("TestId", tags_dict.get("TestId"))

            if node_type == "sct-runner":
                LOGGER.info("Skipping SCT Runner instance '%s'", instance_id)
                continue

            if keep_action not in ("terminate", ""):
                LOGGER.info(
                    "Post behavior keeps OCI instance '%s' [name=%s] running/stopped. keep_action=%s",
                    instance_id,
                    display_name,
                    keep_action,
                )
                if instance.lifecycle_state == "RUNNING" and not dry_run:
                    try:
                        compute_client.instance_action(instance_id, "STOP")
                        LOGGER.info("Stopped OCI instance %s (id: %s)", display_name, instance_id)
                    except Exception as e:  # noqa: BLE001
                        LOGGER.error("Failed to stop OCI instance %s (id: %s): %s", display_name, instance_id, e)
                continue

            LOGGER.info(
                "Going to terminate OCI instance '%s' [name=%s] in region %s", instance_id, display_name, region
            )
            if dry_run:
                continue
            try:
                compute_client.terminate_instance(instance_id)
                argus_client = init_argus_client(test_id)
                terminate_resource_in_argus(client=argus_client, resource_name=display_name)
                LOGGER.info("Terminated OCI instance %s (id: %s)", display_name, instance_id)
            except Exception as e:  # noqa: BLE001
                LOGGER.error("Failed to terminate OCI instance %s (id: %s): %s", display_name, instance_id, e)


def clean_orphan_block_volumes_oci(tags_dict: dict, dry_run: bool = False, regions: list[str] | None = None) -> None:
    """Clean orphan OCI block volumes created for non-dense shapes.

    Targets only AVAILABLE volumes with names ending in "-vol-<N>".
    Cleanup respects `keep_action` tag (`terminate`/empty => delete, anything else => keep).
    """
    if not tags_dict:
        LOGGER.error("tags_dict not provided (can't clean all OCI volumes)")
        return

    compartment_id = get_oci_compartment_id()
    regions = regions or SUPPORTED_REGIONS
    for region in regions:
        block_storage_client = OciService().get_block_storage_client(region=region)
        try:
            volumes = list(
                oci.pagination.list_call_get_all_results_generator(
                    block_storage_client.list_volumes,
                    yield_mode="record",
                    compartment_id=compartment_id,
                    lifecycle_state="AVAILABLE",
                )
            )
        except Exception as exc:  # noqa: BLE001
            LOGGER.warning("Failed to list OCI block volumes in region %s: %s", region, exc)
            continue

        if not volumes:
            continue

        for volume in volumes:
            volume_name = volume.display_name or ""
            if "-vol-" not in volume_name:
                continue

            volume_tags = (volume.defined_tags or {}).get(TAG_NAMESPACE, {})
            volume_matches_tags = _oci_tags_match(volume_tags, tags_dict) if volume_tags else False

            if not volume_matches_tags:
                continue

            keep_action = oci_keep_action(volume_tags)
            if keep_action not in ("terminate", ""):
                LOGGER.info(
                    "Post behavior keeps OCI block volume '%s' in region %s. keep_action=%s",
                    volume.id,
                    region,
                    keep_action,
                )
                continue

            LOGGER.info(
                "Going to delete OCI block volume '%s' [name=%s] in region %s",
                volume.id,
                volume_name,
                region,
            )

            if dry_run:
                continue
            delete_oci_volume_with_retry(
                block_storage_client=block_storage_client,
                volume_id=volume.id,
                volume_name=volume_name,
                region=region,
                logger=LOGGER,
            )


def clean_instances_aws(tags_dict: dict, regions=None, dry_run=False):
    """Remove all instances with specific tags in AWS."""

    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    if regions:
        aws_instances = {}
        for region in regions:
            aws_instances |= list_instances_aws(tags_dict=tags_dict, region_name=region, group_as_region=True)
    else:
        aws_instances = list_instances_aws(tags_dict=tags_dict, group_as_region=True)

    for region, instance_list in aws_instances.items():
        if not instance_list:
            LOGGER.info("There are no instances to remove in AWS region %s", region)
            continue
        client: EC2Client = boto3.client("ec2", region_name=region)
        for instance in instance_list:
            tags = aws_tags_to_dict(instance.get("Tags"))
            name = tags.get("Name", "N/A")
            node_type = tags.get("NodeType")
            instance_id = instance["InstanceId"]
            if node_type and node_type == "sct-runner":
                LOGGER.info("Skipping Sct Runner instance '%s'", instance_id)
                continue
            LOGGER.info(f"Going to delete '{instance_id}' [name={name}] ")
            if not dry_run:
                response = client.terminate_instances(InstanceIds=[instance_id])
                argus_client = init_argus_client(tags_dict.get("TestId"))
                terminate_resource_in_argus(client=argus_client, resource_name=name)
                LOGGER.debug("Done. Result: %s\n", response["TerminatingInstances"])


def clean_elastic_ips_aws(tags_dict, regions=None, dry_run=False):
    """
    Remove all elastic ips with specific tags AWS

    :param tags_dict: key-value pairs used for filtering
    :param regions: list of the AWS regions to consider
    :return: None
    """
    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    if regions:
        aws_instances = {}
        for region in regions:
            aws_instances |= list_elastic_ips_aws(tags_dict=tags_dict, region_name=region, group_as_region=True)
    else:
        aws_instances = list_elastic_ips_aws(tags_dict=tags_dict, group_as_region=True)

    for region, eip_list in aws_instances.items():
        if not eip_list:
            LOGGER.info("There are no EIPs to remove in AWS region %s", region)
            continue
        client: EC2Client = boto3.client("ec2", region_name=region)
        for eip in eip_list:
            association_id = eip.get("AssociationId")
            if association_id and not dry_run:
                response = client.disassociate_address(AssociationId=association_id)
                LOGGER.debug("disassociate_address. Result: %s\n", response)
            allocation_id = eip["AllocationId"]
            LOGGER.info("Going to release '%s' [public_ip={%s}]", allocation_id, eip["PublicIp"])
            if not dry_run:
                response = client.release_address(AllocationId=allocation_id)
                LOGGER.debug("Done. Result: %s\n", response)


def clean_test_security_groups(tags_dict, regions=None, dry_run=False):
    """
    Remove all security groups with specific tags AWS

    :param tags_dict: key-value pairs used for filtering
    :param regions: list of the AWS regions to consider
    :return: None
    """
    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    if regions:
        aws_instances = {}
        for region in regions:
            aws_instances |= list_test_security_groups(tags_dict=tags_dict, region_name=region, group_as_region=True)
    else:
        aws_instances = list_test_security_groups(tags_dict=tags_dict, group_as_region=True)

    for region, sg_list in aws_instances.items():
        if not sg_list:
            LOGGER.info("There are no SGs to remove in AWS region %s", region)
            continue
        client: EC2Client = boto3.client("ec2", region_name=region)
        for security_group in sg_list:
            group_id = security_group.get("GroupId")
            LOGGER.info("Going to delete '%s'", group_id)
            if not dry_run:
                try:
                    response = client.delete_security_group(GroupId=group_id)
                    LOGGER.debug("Done. Result: %s\n", response)
                except Exception as ex:  # noqa: BLE001
                    LOGGER.debug("Failed with: %s", str(ex))


def clean_aws_kms_alias(tags_dict, region_names):
    # NOTE: try to delete KMS key alias which could be created by the AWS-KMS nemesis
    test_id = tags_dict.get("TestId", "TestIdNotFound")
    if any(("db" in node_type for node_type in tags_dict.get("NodeType", []))):
        AwsKms(region_names=region_names).delete_alias(f"alias/testid-{test_id}", tolerate_errors=True)
    else:
        LOGGER.info("Skip AWS KMS alias deletion because DB nodes deletion was not scheduled")


def clean_load_balancers_aws(tags_dict, regions=None, dry_run=False):
    """
    Remove all load balancers with specific tags AWS

    :param tags_dict: key-value pairs used for filtering
    :param regions: list of the AWS regions to consider
    :param dry_run: if True, wouldn't delete any resource
    :return: None
    """

    # don't if `post_behavior_k8s_cluster` was set to not destroy
    if "NodeType" in tags_dict and "k8s" not in tags_dict.get("NodeType"):
        return

    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    elbs_per_region = list_load_balancers_aws(tags_dict=tags_dict, regions=regions, group_as_region=True)

    for region, elb_list in elbs_per_region.items():
        if not elb_list:
            LOGGER.info("There are no ELBs to remove in AWS region %s", region)
            continue
        client = boto3.client("elb", region_name=region)
        for elb in elb_list:
            arn = elb.get("ResourceARN")
            LOGGER.info("Going to delete '%s'", arn)
            if not dry_run:
                try:
                    response = client.delete_load_balancer(LoadBalancerName=arn.split("/")[1])
                    LOGGER.debug("Done. Result: %s\n", response)
                except Exception as ex:  # noqa: BLE001
                    LOGGER.debug("Failed with: %s", str(ex))


def clean_cloudformation_stacks_aws(tags_dict, regions=None, dry_run=False):
    """
    Remove all cloudformation stacks with specific tags AWS

    :param tags_dict: key-value pairs used for filtering
    :param regions: list of the AWS regions to consider
    :param dry_run: if True, wouldn't delete any resource
    :return: None
    """

    # don't if `post_behavior_k8s_cluster` was set to not destroy
    if "NodeType" in tags_dict and "k8s" not in tags_dict.get("NodeType"):
        return

    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    stacks_per_region = list_cloudformation_stacks_aws(tags_dict=tags_dict, regions=regions, group_as_region=True)

    for region, stacks_list in stacks_per_region.items():
        if not stacks_list:
            LOGGER.info("There are no cloudformation stacks to remove in AWS region %s", region)
            continue
        client = boto3.client("cloudformation", region_name=region)
        for stack in stacks_list:
            arn = stack.get("ResourceARN")
            LOGGER.info("Going to delete '%s'", arn)
            if not dry_run:
                try:
                    response = client.delete_stack(StackName=arn.split("/")[1])
                    LOGGER.debug("Done. Result: %s\n", response)
                except Exception as ex:  # noqa: BLE001
                    LOGGER.debug("Failed with: %s", str(ex))


def clean_launch_templates_aws(tags_dict, regions=None, dry_run=False):
    """
    Remove all VM launch templates with specific tags.

    :param tags_dict: key-value pairs used for filtering
    :param regions: list of the AWS regions to consider
    :param dry_run: if True, wouldn't delete any resource
    :return: None
    """

    assert tags_dict, "Can't cleanup launch templates because 'tags_dict' was not provided."

    lts_per_region = list_launch_templates_aws(tags_dict=tags_dict, regions=regions)
    for region_name, lt_list in lts_per_region.items():
        if not lt_list:
            LOGGER.info("There are no LaunchTemplates to remove in AWS region %s", region_name)
            continue
        ec2_client = boto3.client("ec2", region_name=region_name)
        for current_lt in lt_list:
            current_lt_name = current_lt.get("LaunchTemplateName")
            current_lt_id = current_lt.get("LaunchTemplateId")
            LOGGER.info("Going to delete LaunchTemplate, ID: '%s', Name: '%s'", current_lt_id, current_lt_name)
            if dry_run:
                continue
            try:
                if current_lt_name:
                    deletion_args = {"LaunchTemplateName": current_lt_name}
                else:
                    deletion_args = {"LaunchTemplateId": current_lt_id}
                response = ec2_client.delete_launch_template(**deletion_args)
                LOGGER.info(
                    "Successfully deleted '%s' LaunchTemplate. Response: %s\n",
                    (current_lt_name or current_lt_id),
                    response,
                )
            except Exception as ex:  # noqa: BLE001
                LOGGER.info("Failed to delete the '%s' LaunchTemplate: %s", deletion_args, str(ex))


def clean_instances_gce(tags_dict: dict, dry_run=False):
    """
    Remove all instances with specific tags GCE

    :param tags_dict: key-value pairs used for filtering
    :param dry_run: if True, only log what would be deleted without actually deleting
    :return: None
    """
    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    gce_instances_to_clean = list_instances_gce(tags_dict=tags_dict)

    if not gce_instances_to_clean:
        LOGGER.info("There are no GCE instances to remove in the %s project", os.environ.get("SCT_GCE_PROJECT"))
        return

    def delete_instance(instance_with_tags: tuple[GceInstance, dict]):
        instance, tags_dict = instance_with_tags
        LOGGER.info("Going to delete: %s (%s project)", instance.name, os.environ.get("SCT_GCE_PROJECT"))

        if not dry_run:
            instances_client, info = get_gce_compute_instances_client()
            res = instances_client.delete(
                instance=instance.name, project=info["project_id"], zone=instance.zone.split("/")[-1]
            )
            res.done()
            argus_client = init_argus_client(tags_dict.get("TestId"))
            terminate_resource_in_argus(client=argus_client, resource_name=instance.name)
            LOGGER.info("%s deleted=%s", instance.name, res)

    ParallelObject(map(lambda i: (i, tags_dict), gce_instances_to_clean), timeout=60).run(
        delete_instance, ignore_exceptions=False
    )


def clean_instances_azure(tags_dict: dict, regions=None, dry_run=False):
    """
    Cleans instances by tags.

    :param tags_dict: key-value pairs used for filtering
    :return: None
    """
    assert tags_dict, "Running clean instances without tags would remove all SCT related resources in all regions"

    provisioners = AzureProvisioner.discover_regions(tags_dict.get("TestId", ""), regions=regions)
    for provisioner in provisioners:
        all_instances = provisioner.list_instances()
        instances_to_clean = []
        for instance in all_instances:
            tags = instance.tags
            for tag_k, tag_v in tags_dict.items():
                if tag_k not in tags or (tags[tag_k] not in tag_v if isinstance(tag_v, list) else tags[tag_k] != tag_v):
                    break
            else:
                instances_to_clean.append(instance)
        if len(all_instances) == len(instances_to_clean):
            LOGGER.info("Cleaning everything for test id: %s in region: %s", provisioner.test_id, provisioner.region)
            if not dry_run:
                provisioner.cleanup(wait=False)
                argus_client = init_argus_client(tags_dict.get("TestId"))
                for instance in instances_to_clean:
                    terminate_resource_in_argus(client=argus_client, resource_name=instance.name)
        else:
            LOGGER.info(
                "test id %s from %s - instances to clean: %s",
                provisioner.test_id,
                provisioner.region,
                [inst.name for inst in instances_to_clean],
            )
            if not dry_run:
                for instance in instances_to_clean:
                    instance.terminate(wait=False)


def clean_clusters_gke(tags_dict: dict, dry_run: bool = False) -> None:
    if "NodeType" in tags_dict and "k8s" not in tags_dict.get("NodeType"):
        return
    assert tags_dict, "tags_dict not provided (can't clean all clusters)"
    gke_clusters_to_clean = list_clusters_gke(tags_dict=tags_dict)

    if not gke_clusters_to_clean:
        LOGGER.info("There are no GKE clusters to remove in the %s project", os.environ.get("SCT_GCE_PROJECT"))
        return

    def delete_cluster(cluster):
        if not dry_run:
            LOGGER.info(
                "Going to delete %s GKE cluster from the %s project", cluster.name, os.environ.get("SCT_GCE_PROJECT")
            )
            try:
                res = cluster.destroy()
                LOGGER.info("%s deleted=%s", cluster.name, res)
            except Exception as exc:  # noqa: BLE001
                LOGGER.error(exc)

    ParallelObject(gke_clusters_to_clean, timeout=180).run(delete_cluster, ignore_exceptions=True)


def clean_orphaned_gke_disks(tags_dict: dict, dry_run: bool = False) -> None:
    if "NodeType" in tags_dict and "k8s" not in tags_dict.get("NodeType"):
        return
    try:
        gke_cleaner = GkeCleaner()
        orphaned_disks = gke_cleaner.list_orphaned_gke_disks()
        LOGGER.info(
            "Found following orphaned GKE disks in the %s project: %s",
            os.environ.get("SCT_GCE_PROJECT"),
            orphaned_disks,
        )
        if not dry_run:
            for zone, disk_names in orphaned_disks.items():
                gke_cleaner.clean_disks(disk_names=disk_names, zone=zone)
                LOGGER.info(
                    "Deleted following orphaned GKE disks in the '%s' zone (%s project): %s",
                    zone,
                    os.environ.get("SCT_GCE_PROJECT"),
                    disk_names,
                )
    except Exception as exc:  # noqa: BLE001
        LOGGER.error(exc)


def clean_clusters_eks(tags_dict: dict, regions: list = None, dry_run: bool = False) -> None:
    if "NodeType" in tags_dict and "k8s" not in tags_dict.get("NodeType"):
        return
    assert tags_dict, "tags_dict not provided (can't clean all clusters)"
    eks_clusters_to_clean = list_clusters_eks(tags_dict=tags_dict, regions=regions)

    if not eks_clusters_to_clean:
        LOGGER.info("There are no EKS clusters to remove in %s region(s)" % ",".join(regions) or "all")
        return

    def delete_cluster(cluster):
        if not dry_run:
            LOGGER.info("Going to delete '%s' EKS cluster in the '%s' region.", cluster.name, cluster.region_name)
            try:
                res = cluster.destroy()
                LOGGER.info(
                    "'%s' EKS cluster in the '%s' region has been deleted. Response=%s",
                    cluster.name,
                    cluster.region_name,
                    res,
                )
            except Exception as exc:  # noqa: BLE001
                LOGGER.error(exc)

    ParallelObject(eks_clusters_to_clean, timeout=180).run(delete_cluster, ignore_exceptions=True)


def clean_emr_clusters(tags_dict: dict, regions: list = None, dry_run: bool = False) -> list[tuple[str, Exception]]:
    """Discover and terminate EMR clusters matching the given tags.

    Args:
        tags_dict: Dictionary of tags to filter resources (must include TestId or RunByUser).
        regions: List of AWS regions to search. If None, searches all regions.
        dry_run: If True, only log what would be deleted without terminating.
    """
    assert tags_dict, "tags_dict not provided (can't clean all EMR clusters)"
    regions = regions or all_aws_regions()

    def clean_region(region):
        matching_clusters = list_emr_clusters(tags_dict=tags_dict, region_name=region)
        if not matching_clusters:
            LOGGER.info("No EMR clusters to clean up in %s", region)
            return

        emr_client = boto3.client("emr", region_name=region)
        for cluster_info in matching_clusters:
            cluster_id = cluster_info["ClusterId"]
            cluster_name = cluster_info.get("Name", "N/A")
            LOGGER.info(
                "Going to terminate EMR cluster '%s' (ID: %s) in %s [state: %s]",
                cluster_name,
                cluster_id,
                region,
                cluster_info.get("State", "unknown"),
            )
            if not dry_run:
                try:
                    emr_client.terminate_job_flows(JobFlowIds=[cluster_id])
                    LOGGER.info("EMR cluster %s termination initiated", cluster_id)
                except Exception as exc:  # noqa: BLE001
                    LOGGER.error("Failed to terminate EMR cluster %s: %s", cluster_id, exc)

    return run_per_region_ignore_failures(regions, clean_region, "EMR clusters", timeout=120)


def clean_resources_according_post_behavior(params, config, logdir, dry_run=False):
    critical_events = get_testrun_status(params.get("TestId"), logdir, only_critical=True)
    actions_per_type = get_post_behavior_actions(config)
    LOGGER.debug(actions_per_type)

    node_types_to_cleanup = []
    for cluster_nodes_type, action_type in actions_per_type.items():
        if action_type["action"] == "keep":
            LOGGER.info("Post behavior %s for %s. Keep resources running", action_type["action"], cluster_nodes_type)
        elif action_type["action"] == "destroy":
            LOGGER.info("Post behavior %s for %s. Schedule cleanup", action_type["action"], cluster_nodes_type)
            for node_type in action_type["node_types"]:
                node_types_to_cleanup.append(node_type)
            continue
        elif action_type["action"] == "keep-on-failure" and not critical_events:
            LOGGER.info(
                "Post behavior %s for %s. No critical events found. Schedule cleanup.",
                action_type["action"],
                cluster_nodes_type,
            )
            for node_type in action_type["node_types"]:
                node_types_to_cleanup.append(node_type)
            continue
        else:
            LOGGER.info(
                "Post behavior %s for %s. Test run Failed. Keep resources running",
                action_type["action"],
                cluster_nodes_type,
            )
            continue
    clean_cloud_resources(params | {"NodeType": node_types_to_cleanup}, config=config, dry_run=dry_run)


@retrying(n=30, sleep_time=10, allowed_exceptions=(ClientError,))
def clean_placement_groups_aws(tags_dict: dict, regions=None, dry_run=False):
    """Remove all placement groups with specific tags in AWS."""
    tags_dict = {key: value for key, value in tags_dict.items() if key != "NodeType"}
    assert tags_dict, "tags_dict not provided (can't clean all instances)"
    if regions:
        aws_placement_groups = {}
        for region in regions:
            aws_placement_groups |= list_placement_groups_aws(
                tags_dict=tags_dict, region_name=region, group_as_region=True
            )
    else:
        aws_placement_groups = list_placement_groups_aws(tags_dict=tags_dict, group_as_region=True)

    for region, instance_list in aws_placement_groups.items():
        if not instance_list:
            LOGGER.debug("There are no placement groups to remove in AWS region %s", region)
            continue
        client: EC2Client = boto3.client("ec2", region_name=region)
        for instance in instance_list:
            name = instance.get("GroupName")
            LOGGER.info(f"Going to delete placement group '{name} ")
            if not dry_run:
                try:
                    response = client.delete_placement_group(GroupName=name)
                    LOGGER.info("Placement group deleted: %s\n", response)
                except Exception as ex:
                    LOGGER.debug("Failed to delete placement group: %s", str(ex))
                    raise


def cleanup_cluster_vpc_peering(
    api_client: ScyllaCloudAPIClient, account_id: str, cluster_id: int, cluster_name: str, config: dict
) -> None:
    try:
        peerings = api_client.get_vpc_peers(account_id=account_id, cluster_id=cluster_id)
    except ScyllaCloudAPIError as e:
        if "041104" not in str(e):  # '041104' is the "The cluster does not have VPC peering enabled" Siren error code
            LOGGER.error("Failed to cleanup Scylla Cloud VPC peering for cluster %s: %s", cluster_name, e)
        return

    xcloud_provider = config.get("xcloud_provider").lower()
    region_name = config.cloud_provider_params.get("region")
    for peering in peerings:
        peering_id = peering.get("id")
        LOGGER.info("Deleting Scylla Cloud VPC peering %s for cluster %s", peering_id, cluster_name)

        peering_details = api_client.get_vpc_peer_details(
            account_id=account_id, cluster_id=cluster_id, peer_id=peering_id
        )
        cloud_vpc_cidr = api_client.get_cluster_details(account_id=account_id, cluster_id=cluster_id, enriched=True)[
            "dc"
        ]["cidrBlock"]

        try:
            if xcloud_provider == "aws":
                _cleanup_aws_vpc_peering_configuration(region_name, cloud_vpc_cidr, peering_details.get("externalId"))
            elif xcloud_provider == "gce":
                _cleanup_gcp_vpc_peering_configuration(region_name, peering_id, peering_details.get("projectId"))
        except Exception as e:  # noqa: BLE001
            LOGGER.error(
                "Failed to cleanup %s side configuration for Scylla Cloud peering %s: %s",
                xcloud_provider.upper(),
                peering_id,
                e,
            )

        api_client.delete_vpc_peer(account_id=account_id, cluster_id=cluster_id, peer_id=peering_id)


def _cleanup_aws_vpc_peering_configuration(region_name: str, cloud_vpc_cidr: str, peering_id: str) -> None:
    aws_region = AwsRegion(region_name)
    for route_table in aws_region.sct_route_tables:
        for route in route_table.routes_attribute:
            if (
                route.get("DestinationCidrBlock") == cloud_vpc_cidr
                and route.get("VpcPeeringConnectionId") == peering_id
            ):
                try:
                    LOGGER.info(
                        "Removing route: %s -> %s from route table %s", cloud_vpc_cidr, peering_id, route_table.id
                    )
                    aws_region.client.delete_route(RouteTableId=route_table.id, DestinationCidrBlock=cloud_vpc_cidr)
                except Exception as e:  # noqa: BLE001
                    if "InvalidRoute.NotFound" not in str(e):
                        LOGGER.warning("Failed to remove route for %s: %s", cloud_vpc_cidr, e)


def _cleanup_gcp_vpc_peering_configuration(region_name: str, peering_id: str, peer_project: str) -> None:
    gce_region = GceRegion(region_name)
    peering_cleanup_success = gce_region.cleanup_vpc_peering_connection(
        f"sct-to-scylla-cloud-{peer_project}-{peering_id}"
    )
    if not peering_cleanup_success:
        LOGGER.error("Failed to clean up GCP side network peering %s", peering_id)


def clean_blackhole_routes_aws(config: dict, regions=None, dry_run=False) -> None:
    cidr_pool = ipaddress.ip_network(config.get("xcloud_vpc_peering").get("cidr_pool_base"))
    regions = regions or [config.cloud_provider_params.get("region")]
    for region_name in regions:
        aws_region = AwsRegion(region_name)
        cleaned_routes = []

        for index in [None, 0, 1]:
            route_table = aws_region.sct_route_table(index=index)
            if not route_table:
                continue

            for route in route_table.routes_attribute:
                dest_cidr = route.get("DestinationCidrBlock")
                route_state = route.get("State")

                if not dest_cidr or route_state != "blackhole":
                    continue

                route_network = ipaddress.ip_network(dest_cidr)
                if route_network.subnet_of(cidr_pool):
                    LOGGER.info(
                        "Removing blackhole route: %s from route table %s in region %s",
                        dest_cidr,
                        route_table.id,
                        region_name,
                    )
                    if not dry_run:
                        aws_region.client.delete_route(RouteTableId=route_table.id, DestinationCidrBlock=dest_cidr)
                        cleaned_routes.append(dest_cidr)

        if cleaned_routes:
            LOGGER.info("Cleaned %s blackhole routes in region %s", cleaned_routes, region_name)


def clean_inactive_peerings_gce(config: dict, regions=None, dry_run=False) -> None:
    regions = regions or [config.cloud_provider_params.get("region")]
    for region_name in regions:
        gce_region = GceRegion(region_name)
        cleaned_peerings = []

        network = gce_region.network
        if not network.peerings:
            continue

        for peering in network.peerings:
            if peering.state != "INACTIVE":
                continue

            peer_project = peering.network.split("/")[-4]
            LOGGER.info(
                "Removing inactive peering %s to %s Scylla Cloud project, in region %s",
                peering.name,
                peer_project,
                region_name,
            )
            if not dry_run:
                # retry with exponential backoff if another peering operation is in progress
                max_retries, retry_delay = 5, 3
                for attempt in range(max_retries):
                    try:
                        gce_region.cleanup_vpc_peering_connection(peering.name)
                        cleaned_peerings.append(peering.name)
                        break
                    except google.api_core.exceptions.BadRequest as exc:
                        if "peering operation in progress" in str(exc).lower() and attempt < max_retries - 1:
                            LOGGER.warning(
                                "Peering operation is in progress, waiting %s seconds before retry (attempt %s/%s)",
                                retry_delay,
                                attempt + 1,
                                max_retries,
                            )
                            time.sleep(retry_delay)
                            retry_delay *= 2
                        else:
                            LOGGER.error(
                                "Failed to remove peering %s after %s attempts: %s", peering.name, attempt + 1, exc
                            )
                            raise

        if cleaned_peerings:
            LOGGER.info("Cleaned %s inactive peerings in region %s", cleaned_peerings, region_name)


def clean_clusters_scylla_cloud(tags_dict: dict, config: dict, dry_run: bool = False) -> None:
    """Clean up Scylla Cloud resources based on tags"""
    assert tags_dict, "tags_dict not provided (can't clean all instances)"

    if "NodeType" in tags_dict and "scylla-db" not in tags_dict.get("NodeType"):
        return

    if dry_run:
        LOGGER.info("DRY RUN: No actual resources will be deleted")

    api_client = ScyllaCloudAPIClient(
        api_url=config.cloud_env_credentials["base_url"], auth_token=config.cloud_env_credentials["api_token"]
    )

    account_id = api_client.get_account_details().get("accountId")
    clusters = api_client.get_clusters(account_id=account_id, enriched=True)
    LOGGER.info("Found %s cluster(s) in Scylla Cloud account", len(clusters))

    clusters_to_delete = []
    test_id = tags_dict.get("TestId")
    # TODO: uncomment after adding tags/metadata to a cloud cluster is implemented
    # run_by_user = tags_dict.get('RunByUser', '').replace('.', '-')
    for cluster_data in clusters:
        cluster_name = cluster_data.get("clusterName", "")
        cluster_id = cluster_data.get("id")

        should_delete = False
        if test_id and str(test_id)[:8] in cluster_name:
            should_delete = True
        # TODO: uncomment after adding tags/metadata to a cloud cluster is implemented
        # elif run_by_user and run_by_user in cluster_name:
        #     should_delete = True

        if should_delete:
            clusters_to_delete.append({"id": cluster_id, "name": cluster_name})

    if not clusters_to_delete:
        LOGGER.info("No clusters found matching the cleanup criteria")
        return

    LOGGER.info("Found %s cluster(s) to delete", len(clusters_to_delete))

    for cluster in clusters_to_delete:
        cluster_id, cluster_name = cluster["id"], cluster["name"]
        LOGGER.info("Going to delete cluster %s (ID: %s)", cluster_name, cluster_id)

        if not dry_run:
            try:
                cleanup_cluster_vpc_peering(api_client, account_id, cluster_id, cluster_name, config)
                api_client.delete_cluster(account_id=account_id, cluster_id=cluster_id, cluster_name=cluster_name)

                argus_client = init_argus_client(tags_dict.get("TestId"))
                terminate_resource_in_argus(client=argus_client, resource_name=cluster_name)
            except Exception as e:  # noqa: BLE001
                LOGGER.error(f"Failed to delete cluster {cluster_name}: {e}")

    xcloud_provider = config.get("xcloud_provider")
    if xcloud_provider == "aws":
        aws_regions = config.region_names
        clean_instances_aws(tags_dict, regions=aws_regions, dry_run=dry_run)
        clean_elastic_ips_aws(tags_dict, regions=aws_regions, dry_run=dry_run)
        clean_test_security_groups(tags_dict, regions=aws_regions, dry_run=dry_run)
        clean_placement_groups_aws(tags_dict, regions=aws_regions, dry_run=dry_run)
        clean_blackhole_routes_aws(config, regions=aws_regions, dry_run=dry_run)
    elif xcloud_provider == "gce":
        gce_projects = [config.get("gce_project") or "gcp-sct-project-1"]
        gce_regions = config.gce_datacenters
        for project in gce_projects:
            with environment(SCT_GCE_PROJECT=project):
                clean_instances_gce(tags_dict, dry_run=dry_run)
        clean_inactive_peerings_gce(config, regions=gce_regions, dry_run=dry_run)
