<commit_before>"""
AWS Resource Enumerators

Utility to find Drift targets running in the AWS Cloud.
"""
import os
import socket
import logging
import re

import boto3
import requests
from requests.utils import urlparse

from driftconfig.util import get_drift_config


log = logging.getLogger(__name__)


HEALTHCHECK_TIMEOUT = 1.0  # Timeout for target health check ping.
HEALTHCHECK_PORT = 8080  # HTTP server port on targets.


def _get_ec2_targets_from_aws(tier_name):

    conf = get_drift_config(tier_name=tier_name)
    deployables = conf.table_store.get_table('deployables').find({'tier_name': tier_name})
    deployables = {d['deployable_name']: d for d in deployables}  # Turn into a dict

    # Enumerate EC2's that match the entries in 'deployables' and are tagged as 'api-target'.
    if 'aws' not in conf.tier:
        raise RuntimeError("'aws' section missing from tier configuration.")

    ec2_client = boto3.resource('ec2', region_name=conf.tier['aws']['region'])
    filters = {
        'instance-state-name': 'running',
        'tag:tier': conf.tier['tier_name'],
    }
    def filterize(d):
        """
        Return dictionary 'd' as a boto3 "filters" object by unfolding it to a list of
        dict with 'Name' and 'Values' entries.
        """
        return [{'Name': k, 'Values': [v]} for k, v in d.items()]

    log.info("Fetch EC2 instances that match %s", filterize(filters))
    ec2_instances = list(ec2_client.instances.filter(Filters=filterize(filters)))

    # If the instances are part of an autoscaling group, make sure they are healthy and in service.
    region_name = conf.tier['aws']['region']
    vpc = _get_vpc_for_tier(region_name=region_name, tier_name=tier_name)
    autoscaling = boto3.client('autoscaling', region_name=region_name)
    auto_ec2s = autoscaling.describe_auto_scaling_instances(
        InstanceIds=[ec2.instance_id for ec2 in ec2_instances]
    )
    auto_ec2s = {auto_ec2['InstanceId']: auto_ec2 for auto_ec2 in auto_ec2s['AutoScalingInstances']}
    # auto_ec2s is a dict with instance id as key, and value is a dict with LifecycleState and HealthStatus key.

    api_targets = {}
    for ec2 in ec2_instances:
        # Analyse the EC2 instances and see if they are supposed to be a target of the api router.

        def fold_tags(tags, key_name=None, value_name=None):
            """Fold boto3 resource tags array into a dictionary."""
            return {tag['Key']: tag['Value'] for tag in tags}

        tags = fold_tags(ec2.tags)
        api_status = tags.get('api-status')
        api_target = tags.get('api-target')
        api_port = tags.get('api-port')
        name = tags.get('Name')

        # Check if instance is being scaled in or out by autoscaling group.
        if ec2.instance_id in auto_ec2s:
            lifecycle_state = auto_ec2s[ec2.instance_id]['LifecycleState']

            if lifecycle_state.startswith('Terminating'):
                # The instances are normally equipped with a lifecycle hook that specifies
                # 2 minute timeout before they are terminated. The 'LivecycleState' value
                # is "Terminating:Waiting" during this transition.
                # The timeout can be extended by by calling record_lifecycle_action_heartbeat()
                # on the lifecycle hook. Example:
                # boto3.client('autoscaling).record_lifecycle_action_heartbeat(...)
                #
                # To gracefully drain the connections, the instance is marked as 'backup'. This
                # simply removes the instance from the round robin load balancing.
                log.info("EC2 instance %s[%s] terminating. Marking it as 'backup' to drain connections.", name, ec2.instance_id[:7])
                tags['api-param'] = 'backup'  # This will enable connection draining in Nginx.
            elif lifecycle_state != 'InService':
                log.warning("EC2 instance %s[%s] not in service yet: %s", name, ec2.instance_id[:7], lifecycle_state)
                continue

        if ec2.vpc_id != vpc['VpcId']:
            log.warning("EC2 instance %s[%s] not in correct VPC. Fix tier tag!", name, ec2.instance_id[:7])
            continue

        if not any([api_status, api_target, api_port]):
            log.info("EC2 instance %s[%s] not in rotation, as it's not configured as api-target.", name, ec2.instance_id[:7])
            continue

        if any([api_status, api_target, api_port]) and not all([api_status, api_target, api_port]):
            log.warning("EC2 instance %s[%s] must define all api tags, not just some: %s.", name, ec2.instance_id[:7], tags)
            continue

        if not api_port.isnumeric():
            log.warning("EC2 instance %s[%s] has bogus 'api-port' tag: %s.", name, ec2.instance_id[:7], api_port)
            continue

        deployable = deployables.get(api_target)
        if not deployable:
            log.warning("EC2 instance %s[%s]: No deployable defined for api-target '%s'.", name, ec2.instance_id[:7], api_target)
            continue

        if api_status not in ['online', 'online2']:
            log.info("EC2 instance %s[%s] not in rotation, api-status tag is '%s'.", name, ec2.instance_id[:7], api_status)
            continue

        log.info(
            "EC2 instance %s[%s] in rotation. [%s:%s:%s]",
            name, ec2.instance_id[:7], api_target, api_status, api_port
        )

        target = {
            'name': name,
            'image_id': ec2.image_id,
            'instance_id': ec2.id,
            'instance_type': ec2.instance_type,
            'launch_time': ec2.launch_time.isoformat() + 'Z',
            'placement': ec2.placement,
            'private_ip_address': ec2.private_ip_address,
            'public_ip_address': ec2.public_ip_address,
            'state_name': ec2.state['Name'],
            'state_transition_reason': ec2.state_transition_reason,
            'subnet_id': ec2.subnet_id,
            'tags': tags,
            'vpc_id': ec2.vpc_id,
            'comment': "{} [{}] [{}]".format(name, ec2.instance_type, ec2.placement['AvailabilityZone']),
        }

        api_targets.setdefault(api_target, []).append(target)

    return api_targets


def _healthcheck_targets(api_targets):
    """Pings targets in 'api_targets' for health and removes a new map of 'healhty_targets' as a result."""
    healthy_targets = {}

    for api_target_name, targets in api_targets.items():
        healthy_targets[api_target_name] = []
        for target in targets[:]:
            log.info("Checking health of %s", target['private_ip_address'])
            # Ping the target for health. The assumption is that the target runs a plain
            # http server on port 8080 and responds to / with a 200.

            # HACK: There is no uniform healtch check path on targets. Trying out a few until
            # this gets cleaned up:
            # Try /healthcheck
            url = 'http://{}:{}/healthcheck'.format(target['private_ip_address'], HEALTHCHECK_PORT)
            status, message = _check_url(url)
            if status != 200:
                # Try /
                url = 'http://{}:{}/'.format(target['private_ip_address'], HEALTHCHECK_PORT)
                status, message = _check_url(url)
            if status != 200:
                # Try /gurko
                url = 'http://{}:{}/gurko'.format(target['private_ip_address'], HEALTHCHECK_PORT)
                status, message = _check_url(url)

            if status != 200:<commit_msg>Rename /gurko endpoint as /healthcheck<commit_after>"""
AWS Resource Enumerators

Utility to find Drift targets running in the AWS Cloud.
"""
import os
import socket
import logging
import re

import boto3
import requests
from requests.utils import urlparse

from driftconfig.util import get_drift_config


log = logging.getLogger(__name__)


HEALTHCHECK_TIMEOUT = 1.0  # Timeout for target health check ping.
HEALTHCHECK_PORT = 8080  # HTTP server port on targets.


def _get_ec2_targets_from_aws(tier_name):

    conf = get_drift_config(tier_name=tier_name)
    deployables = conf.table_store.get_table('deployables').find({'tier_name': tier_name})
    deployables = {d['deployable_name']: d for d in deployables}  # Turn into a dict

    # Enumerate EC2's that match the entries in 'deployables' and are tagged as 'api-target'.
    if 'aws' not in conf.tier:
        raise RuntimeError("'aws' section missing from tier configuration.")

    ec2_client = boto3.resource('ec2', region_name=conf.tier['aws']['region'])
    filters = {
        'instance-state-name': 'running',
        'tag:tier': conf.tier['tier_name'],
    }
    def filterize(d):
        """
        Return dictionary 'd' as a boto3 "filters" object by unfolding it to a list of
        dict with 'Name' and 'Values' entries.
        """
        return [{'Name': k, 'Values': [v]} for k, v in d.items()]

    log.info("Fetch EC2 instances that match %s", filterize(filters))
    ec2_instances = list(ec2_client.instances.filter(Filters=filterize(filters)))

    # If the instances are part of an autoscaling group, make sure they are healthy and in service.
    region_name = conf.tier['aws']['region']
    vpc = _get_vpc_for_tier(region_name=region_name, tier_name=tier_name)
    autoscaling = boto3.client('autoscaling', region_name=region_name)
    auto_ec2s = autoscaling.describe_auto_scaling_instances(
        InstanceIds=[ec2.instance_id for ec2 in ec2_instances]
    )
    auto_ec2s = {auto_ec2['InstanceId']: auto_ec2 for auto_ec2 in auto_ec2s['AutoScalingInstances']}
    # auto_ec2s is a dict with instance id as key, and value is a dict with LifecycleState and HealthStatus key.

    api_targets = {}
    for ec2 in ec2_instances:
        # Analyse the EC2 instances and see if they are supposed to be a target of the api router.

        def fold_tags(tags, key_name=None, value_name=None):
            """Fold boto3 resource tags array into a dictionary."""
            return {tag['Key']: tag['Value'] for tag in tags}

        tags = fold_tags(ec2.tags)
        api_status = tags.get('api-status')
        api_target = tags.get('api-target')
        api_port = tags.get('api-port')
        name = tags.get('Name')

        # Check if instance is being scaled in or out by autoscaling group.
        if ec2.instance_id in auto_ec2s:
            lifecycle_state = auto_ec2s[ec2.instance_id]['LifecycleState']

            if lifecycle_state.startswith('Terminating'):
                # The instances are normally equipped with a lifecycle hook that specifies
                # 2 minute timeout before they are terminated. The 'LivecycleState' value
                # is "Terminating:Waiting" during this transition.
                # The timeout can be extended by by calling record_lifecycle_action_heartbeat()
                # on the lifecycle hook. Example:
                # boto3.client('autoscaling).record_lifecycle_action_heartbeat(...)
                #
                # To gracefully drain the connections, the instance is marked as 'backup'. This
                # simply removes the instance from the round robin load balancing.
                log.info("EC2 instance %s[%s] terminating. Marking it as 'backup' to drain connections.", name, ec2.instance_id[:7])
                tags['api-param'] = 'backup'  # This will enable connection draining in Nginx.
            elif lifecycle_state != 'InService':
                log.warning("EC2 instance %s[%s] not in service yet: %s", name, ec2.instance_id[:7], lifecycle_state)
                continue

        if ec2.vpc_id != vpc['VpcId']:
            log.warning("EC2 instance %s[%s] not in correct VPC. Fix tier tag!", name, ec2.instance_id[:7])
            continue

        if not any([api_status, api_target, api_port]):
            log.info("EC2 instance %s[%s] not in rotation, as it's not configured as api-target.", name, ec2.instance_id[:7])
            continue

        if any([api_status, api_target, api_port]) and not all([api_status, api_target, api_port]):
            log.warning("EC2 instance %s[%s] must define all api tags, not just some: %s.", name, ec2.instance_id[:7], tags)
            continue

        if not api_port.isnumeric():
            log.warning("EC2 instance %s[%s] has bogus 'api-port' tag: %s.", name, ec2.instance_id[:7], api_port)
            continue

        deployable = deployables.get(api_target)
        if not deployable:
            log.warning("EC2 instance %s[%s]: No deployable defined for api-target '%s'.", name, ec2.instance_id[:7], api_target)
            continue

        if api_status not in ['online', 'online2']:
            log.info("EC2 instance %s[%s] not in rotation, api-status tag is '%s'.", name, ec2.instance_id[:7], api_status)
            continue

        log.info(
            "EC2 instance %s[%s] in rotation. [%s:%s:%s]",
            name, ec2.instance_id[:7], api_target, api_status, api_port
        )

        target = {
            'name': name,
            'image_id': ec2.image_id,
            'instance_id': ec2.id,
            'instance_type': ec2.instance_type,
            'launch_time': ec2.launch_time.isoformat() + 'Z',
            'placement': ec2.placement,
            'private_ip_address': ec2.private_ip_address,
            'public_ip_address': ec2.public_ip_address,
            'state_name': ec2.state['Name'],
            'state_transition_reason': ec2.state_transition_reason,
            'subnet_id': ec2.subnet_id,
            'tags': tags,
            'vpc_id': ec2.vpc_id,
            'comment': "{} [{}] [{}]".format(name, ec2.instance_type, ec2.placement['AvailabilityZone']),
        }

        api_targets.setdefault(api_target, []).append(target)

    return api_targets


def _healthcheck_targets(api_targets):
    """Pings targets in 'api_targets' for health and removes a new map of 'healhty_targets' as a result."""
    healthy_targets = {}

    for api_target_name, targets in api_targets.items():
        healthy_targets[api_target_name] = []
        for target in targets[:]:
            log.info("Checking health of %s", target['private_ip_address'])
            # Ping the target for health. The assumption is that the target runs a plain
            # http server on port 8080 and responds to / with a 200.

            # HACK: There is no uniform healtch check path on targets. Trying out a few until
            # this gets cleaned up:
            # Try /healthcheck
            url = 'http://{}:{}/healthcheck'.format(target['private_ip_address'], HEALTHCHECK_PORT)
            status, message = _check_url(url)
            if status != 200:
                # Try /
                url = 'http://{}:{}/'.format(target['private_ip_address'], HEALTHCHECK_PORT)
                status, message = _check_url(url)

            if status != 200:
{'fraction_non_alphanumeric': 0.1098566792861006, 'fraction_numerical': 0.012777176852352624, 'mean_word_length': 2.621297429620563, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3478588', 'n_tokens_mistral': 4586, 'n_tokens_neox': 4344, 'n_words': 1284}