<commit_before>import logging
import json

from azure.monitor import MonitorClient
from azure.monitor.models import EventData
from copy import deepcopy
from datetime import datetime, timedelta
from threading import RLock, Condition
from typing import List, Tuple, MutableMapping

import pytz
from abc import ABC
from azure.mgmt.compute import ComputeManagementClient
from azure.mgmt.compute.models import VirtualMachineScaleSet, Sku

from autoscaler.utils import Future

logger = logging.getLogger(__name__)


PRIORITY_TAG = 'priority'


class AzureScaleSet:
    def __init__(self, location: str, resource_group: str, name: str, instance_type: str, capacity: int,
                 provisioning_state: str, timeout_until: datetime = None, timeout_reason: str = None, priority: int = None) -> None:
        self.name = name
        self.instance_type = instance_type
        self.capacity = capacity
        self.provisioning_state = provisioning_state
        self.resource_group = resource_group
        self.location = location
        self.timeout_until = timeout_until
        self.timeout_reason = timeout_reason
        self.priority = priority

    def __str__(self):
        return 'AzureScaleSet({}, {}, {}, {})'.format(self.name, self.instance_type, self.capacity, self.provisioning_state)

    def __repr__(self):
        return str(self)

    def _key(self):
        return (self.name, self.instance_type, self.capacity, self.provisioning_state, self.resource_group, self.location,
                self.timeout_until, self.timeout_reason, self.priority)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, AzureScaleSet):
            return False
        return self._key() == other._key()

    def __hash__(self) -> int:
        return hash(self._key())


class AzureScaleSetInstance:
    def __init__(self, instance_id: str, vm_id: str, launch_time: datetime) -> None:
        self.instance_id = instance_id
        self.vm_id = vm_id
        self.launch_time = launch_time

    def __str__(self):
        return 'AzureScaleSetInstance({}, {}, {})'.format(self.instance_id, self.vm_id, self.launch_time)

    def __repr__(self):
        return str(self)

    def _key(self):
        return (self.instance_id, self.vm_id, self.launch_time)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, AzureScaleSetInstance):
            return False
        return self._key() == other._key()

    def __hash__(self) -> int:
        return hash(self._key())


class AzureApi(ABC):
    def list_scale_sets(self, resource_group_name: str) -> List[AzureScaleSet]:
        pass

    def list_scale_set_instances(self, scale_set: AzureScaleSet) -> List[AzureScaleSetInstance]:
        pass

    def update_scale_set(self, scale_set: AzureScaleSet, new_capacity: int) -> Future:
        pass

    def terminate_scale_set_instances(self, scale_set: AzureScaleSet, instances: List[AzureScaleSetInstance]) -> Future:
        pass


TIMEOUT_PERIOD = timedelta(minutes=15)


class AzureWrapper(AzureApi):
    def __init__(self, compute_client: ComputeManagementClient, monitor_client: MonitorClient) -> None:
        self._compute_client = compute_client
        self._monitor_client = monitor_client

    def list_scale_sets(self, resource_group_name: str) -> List[AzureScaleSet]:
        fifteen_minutes_ago = datetime.now(pytz.utc) - TIMEOUT_PERIOD
        filter_clause = "eventTimestamp ge '{}' and resourceGroupName eq '{}'".format(fifteen_minutes_ago, resource_group_name)
        select_clause = "status,subStatus,properties,resourceId,eventTimestamp"

        failures_by_scale_set: MutableMapping[str, List[EventData]] = {}
        for log in self._monitor_client.activity_logs.list(filter=filter_clause, select=select_clause):
            if (log.status and log.status.value == 'Failed') or (log.properties and log.properties.get('statusCode') == 'Conflict'):
                failures_by_scale_set.setdefault(log.resource_id, []).append(log)

        result = []
        for scale_set in self._compute_client.virtual_machine_scale_sets.list(resource_group_name):
            failures = sorted(failures_by_scale_set.get(scale_set.id, []), key=lambda x: x.event_timestamp, reverse=True)
            timeout_until = None
            timeout_reason = None
            for failure in failures:
                status_message = json.loads(failure.properties.get('statusMessage', "{}")) if failure.properties else {}
                error_details = status_message.get('error', {})
                if 'message' in error_details:
                    timeout_until = failure.event_timestamp + TIMEOUT_PERIOD
                    timeout_reason = error_details['message']
                    # Stop if we found a message with details
                    break
                if timeout_until is None:
                    timeout_until = failure.event_timestamp + TIMEOUT_PERIOD
                    timeout_reason = failure.sub_status.localized_value

            priority = int(scale_set.tags[PRIORITY_TAG]) if PRIORITY_TAG in scale_set.tags else None

            result.append(AzureScaleSet(scale_set.location, resource_group_name, scale_set.name, scale_set.sku.name,
                                        scale_set.sku.capacity, scale_set.provisioning_state, timeout_until=timeout_until,
                                        timeout_reason=timeout_reason, priority=priority))
        return result

    def list_scale_set_instances(self, scale_set: AzureScaleSet) -> List[AzureScaleSetInstance]:
        result = []
        for instance in self._compute_client.virtual_machine_scale_set_vms.list(scale_set.resource_group, scale_set.name, expand="instanceView"):
            launch_time = datetime.now(pytz.utc)<commit_msg>Skip delete failures when considering Scale Sets to timeout<commit_after>import logging
import json

from azure.monitor import MonitorClient
from azure.monitor.models import EventData
from copy import deepcopy
from datetime import datetime, timedelta
from threading import RLock, Condition
from typing import List, Tuple, MutableMapping

import pytz
from abc import ABC
from azure.mgmt.compute import ComputeManagementClient
from azure.mgmt.compute.models import VirtualMachineScaleSet, Sku

from autoscaler.utils import Future

logger = logging.getLogger(__name__)


PRIORITY_TAG = 'priority'


class AzureScaleSet:
    def __init__(self, location: str, resource_group: str, name: str, instance_type: str, capacity: int,
                 provisioning_state: str, timeout_until: datetime = None, timeout_reason: str = None, priority: int = None) -> None:
        self.name = name
        self.instance_type = instance_type
        self.capacity = capacity
        self.provisioning_state = provisioning_state
        self.resource_group = resource_group
        self.location = location
        self.timeout_until = timeout_until
        self.timeout_reason = timeout_reason
        self.priority = priority

    def __str__(self):
        return 'AzureScaleSet({}, {}, {}, {})'.format(self.name, self.instance_type, self.capacity, self.provisioning_state)

    def __repr__(self):
        return str(self)

    def _key(self):
        return (self.name, self.instance_type, self.capacity, self.provisioning_state, self.resource_group, self.location,
                self.timeout_until, self.timeout_reason, self.priority)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, AzureScaleSet):
            return False
        return self._key() == other._key()

    def __hash__(self) -> int:
        return hash(self._key())


class AzureScaleSetInstance:
    def __init__(self, instance_id: str, vm_id: str, launch_time: datetime) -> None:
        self.instance_id = instance_id
        self.vm_id = vm_id
        self.launch_time = launch_time

    def __str__(self):
        return 'AzureScaleSetInstance({}, {}, {})'.format(self.instance_id, self.vm_id, self.launch_time)

    def __repr__(self):
        return str(self)

    def _key(self):
        return (self.instance_id, self.vm_id, self.launch_time)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, AzureScaleSetInstance):
            return False
        return self._key() == other._key()

    def __hash__(self) -> int:
        return hash(self._key())


class AzureApi(ABC):
    def list_scale_sets(self, resource_group_name: str) -> List[AzureScaleSet]:
        pass

    def list_scale_set_instances(self, scale_set: AzureScaleSet) -> List[AzureScaleSetInstance]:
        pass

    def update_scale_set(self, scale_set: AzureScaleSet, new_capacity: int) -> Future:
        pass

    def terminate_scale_set_instances(self, scale_set: AzureScaleSet, instances: List[AzureScaleSetInstance]) -> Future:
        pass


TIMEOUT_PERIOD = timedelta(minutes=15)


class AzureWrapper(AzureApi):
    def __init__(self, compute_client: ComputeManagementClient, monitor_client: MonitorClient) -> None:
        self._compute_client = compute_client
        self._monitor_client = monitor_client

    def list_scale_sets(self, resource_group_name: str) -> List[AzureScaleSet]:
        fifteen_minutes_ago = datetime.now(pytz.utc) - TIMEOUT_PERIOD
        filter_clause = "eventTimestamp ge '{}' and resourceGroupName eq '{}'".format(fifteen_minutes_ago, resource_group_name)
        select_clause = "authorization,status,subStatus,properties,resourceId,eventTimestamp"

        failures_by_scale_set: MutableMapping[str, List[EventData]] = {}
        for log in self._monitor_client.activity_logs.list(filter=filter_clause, select=select_clause):
            if (log.status and log.status.value == 'Failed') or (log.properties and log.properties.get('statusCode') == 'Conflict'):
                if log.authorization and log.authorization.action and 'delete' in log.authorization.action:
                    continue
                failures_by_scale_set.setdefault(log.resource_id, []).append(log)

        result = []
        for scale_set in self._compute_client.virtual_machine_scale_sets.list(resource_group_name):
            failures = sorted(failures_by_scale_set.get(scale_set.id, []), key=lambda x: x.event_timestamp, reverse=True)
            timeout_until = None
            timeout_reason = None
            for failure in failures:
                status_message = json.loads(failure.properties.get('statusMessage', "{}")) if failure.properties else {}
                error_details = status_message.get('error', {})
                if 'message' in error_details:
                    timeout_until = failure.event_timestamp + TIMEOUT_PERIOD
                    timeout_reason = error_details['message']
                    # Stop if we found a message with details
                    break
                if timeout_until is None:
                    timeout_until = failure.event_timestamp + TIMEOUT_PERIOD
                    timeout_reason = failure.sub_status.localized_value

            priority = int(scale_set.tags[PRIORITY_TAG]) if PRIORITY_TAG in scale_set.tags else None

            result.append(AzureScaleSet(scale_set.location, resource_group_name, scale_set.name, scale_set.sku.name,
                                        scale_set.sku.capacity, scale_set.provisioning_state, timeout_until=timeout_until,
                                        timeout_reason=timeout_reason, priority=priority))
        return result

    def list_scale_set_instances(self, scale_set: AzureScaleSet) -> List[AzureScaleSetInstance]:
        result = []
        for instance in self._compute_client.virtual_machine_scale_set_vms.list(scale_set.resource_group, scale_set.name, expand="instanceView"):
            launch_time = datetime.now(pytz.utc)
{'fraction_non_alphanumeric': 0.08945442165290662, 'fraction_numerical': 0.0003404545067665333, 'mean_word_length': 3.0869565217391304, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5684018', 'n_tokens_mistral': 3326, 'n_tokens_neox': 3243, 'n_words': 744}