<commit_before>import logging
import os
import time
from functools import reduce
from typing import Dict
from typing import List
from typing import Set
from typing import Tuple

import pyinotify
from kazoo.protocol.states import EventType
from kazoo.recipe.watchers import ChildrenWatch
from kazoo.recipe.watchers import DataWatch
from requests.exceptions import RequestException

from paasta_tools.deployd.common import get_marathon_clients_from_config
from paasta_tools.deployd.common import get_service_instances_needing_update
from paasta_tools.deployd.common import PaastaThread
from paasta_tools.deployd.common import rate_limit_instances
from paasta_tools.deployd.common import ServiceInstance
from paasta_tools.long_running_service_tools import AUTOSCALING_ZK_ROOT
from paasta_tools.marathon_tools import DEFAULT_SOA_DIR
from paasta_tools.marathon_tools import deformat_job_id
from paasta_tools.marathon_tools import get_marathon_apps_with_clients
from paasta_tools.mesos_maintenance import get_draining_hosts
from paasta_tools.utils import get_services_for_cluster
from paasta_tools.utils import list_all_instances_for_service
from paasta_tools.utils import load_system_paasta_config
from paasta_tools.utils import PATH_TO_SYSTEM_PAASTA_CONFIG_DIR


class PaastaWatcher(PaastaThread):

    def __init__(self, inbox_q, cluster, config):
        super(PaastaWatcher, self).__init__()
        self.daemon = True
        self.inbox_q = inbox_q
        self.cluster = cluster
        self.config = config
        self.is_ready = False


class AutoscalerWatcher(PaastaWatcher):

    def __init__(self, inbox_q, cluster, config, **kwargs):
        super(AutoscalerWatcher, self).__init__(inbox_q, cluster, config)
        self.zk = kwargs.pop('zookeeper_client')
        self.watchers: Dict[str, PaastaWatcher] = {}

    def watch_folder(self, path):
        """recursive nonsense"""
        if "autoscaling.lock" in path:
            return
        self.log.debug("Adding watch on {}".format(path))
        watcher = ChildrenWatch(self.zk, path, func=self.process_folder_event, send_event=True)
        self.watchers[path] = watcher
        children = watcher._prior_children
        if children and ('instances' in children):
            self.watch_node("{}/instances".format(path))
        elif children:
            for child in children:
                self.watch_folder("{}/{}".format(path, child))

    def watch_node(self, path):
        self.log.debug("Adding watch on {}".format(path))
        DataWatch(self.zk, path, func=self.process_node_event, send_event=True)

    def process_node_event(self, data, stat, event):
        self.log.debug("Node change: {}".format(event))
        if event and (event.type == EventType.CREATED or event.type == EventType.CHANGED):
            service, instance = event.path.split('/')[-3:-1]
            self.log.info("Number of instances changed or autoscaling enabled for first time"
                          " for {}.{}".format(service, instance))
            # https://github.com/python/mypy/issues/2852
            service_instance = ServiceInstance(  # type: ignore
                service=service,
                instance=instance,
                cluster=self.config.get_cluster(),
                bounce_by=int(time.time()),
                bounce_timers=None,
                watcher=type(self).__name__,
                failures=0,
            )
            self.inbox_q.put(service_instance)

    def process_folder_event(self, children, event):
        self.log.debug("Folder change: {}".format(event))
        if event and (event.type == EventType.CHILD):
            fq_children = ["{}/{}".format(event.path, child) for child in children]
            for child in fq_children:
                if child not in self.watchers:
                    self.watch_folder(child)
<commit_msg>make sure paasta deployd watches for new watches when autoscaling is enabled for the first time<commit_after>import logging
import os
import time
from functools import reduce
from typing import Dict
from typing import List
from typing import Set
from typing import Tuple

import pyinotify
from kazoo.protocol.states import EventType
from kazoo.recipe.watchers import ChildrenWatch
from kazoo.recipe.watchers import DataWatch
from requests.exceptions import RequestException

from paasta_tools.deployd.common import get_marathon_clients_from_config
from paasta_tools.deployd.common import get_service_instances_needing_update
from paasta_tools.deployd.common import PaastaThread
from paasta_tools.deployd.common import rate_limit_instances
from paasta_tools.deployd.common import ServiceInstance
from paasta_tools.long_running_service_tools import AUTOSCALING_ZK_ROOT
from paasta_tools.marathon_tools import DEFAULT_SOA_DIR
from paasta_tools.marathon_tools import deformat_job_id
from paasta_tools.marathon_tools import get_marathon_apps_with_clients
from paasta_tools.mesos_maintenance import get_draining_hosts
from paasta_tools.utils import get_services_for_cluster
from paasta_tools.utils import list_all_instances_for_service
from paasta_tools.utils import load_system_paasta_config
from paasta_tools.utils import PATH_TO_SYSTEM_PAASTA_CONFIG_DIR


class PaastaWatcher(PaastaThread):

    def __init__(self, inbox_q, cluster, config):
        super(PaastaWatcher, self).__init__()
        self.daemon = True
        self.inbox_q = inbox_q
        self.cluster = cluster
        self.config = config
        self.is_ready = False


class AutoscalerWatcher(PaastaWatcher):

    def __init__(self, inbox_q, cluster, config, **kwargs):
        super(AutoscalerWatcher, self).__init__(inbox_q, cluster, config)
        self.zk = kwargs.pop('zookeeper_client')
        self.watchers: Dict[str, PaastaWatcher] = {}

    def watch_folder(self, path):
        """recursive nonsense"""
        if "autoscaling.lock" in path:
            return
        self.log.debug("Adding folder watch on {}".format(path))
        watcher = ChildrenWatch(self.zk, path, func=self.process_folder_event, send_event=True)
        self.watchers[path] = watcher
        children = watcher._prior_children
        if children and ('instances' in children):
            self.watch_node("{}/instances".format(path))
        elif children:
            for child in children:
                self.watch_folder("{}/{}".format(path, child))

    def _enqueue_service_instance(self, path):
        service, instance = path.split('/')[-3:-1]
        self.log.info("Number of instances changed or autoscaling enabled for first time"
                      " for {}.{}".format(service, instance))
        # https://github.com/python/mypy/issues/2852
        service_instance = ServiceInstance(  # type: ignore
            service=service,
            instance=instance,
            cluster=self.config.get_cluster(),
            bounce_by=int(time.time()),
            bounce_timers=None,
            watcher=type(self).__name__,
            failures=0,
        )
        self.inbox_q.put(service_instance)

    def watch_node(self, path):
        self.log.debug("Adding node watch on {}".format(path))
        DataWatch(self.zk, path, func=self.process_node_event, send_event=True)
        self._enqueue_service_instance(path)

    def process_node_event(self, data, stat, event):
        self.log.debug("Node change: {}".format(event))
        if event and (event.type == EventType.CREATED or event.type == EventType.CHANGED):
            self._enqueue_service_instance(event.path)

    def process_folder_event(self, children, event):
        self.log.debug("Folder change: {}".format(event))
        if event and (event.type == EventType.CHILD):
            fq_children = ["{}/{}".format(event.path, child) for child in children]
            for child in fq_children:
                if child not in self.watchers:
                    self.watch_folder(child)

{'fraction_non_alphanumeric': 0.08636017272034544, 'fraction_numerical': 0.001778003556007112, 'mean_word_length': 3.3079868708971554, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, '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': '3081762', 'n_tokens_mistral': 2348, 'n_tokens_neox': 2267, 'n_words': 540}