<commit_before>#
# Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
#

import re
import gevent
import kazoo.client
import kazoo.exceptions
import kazoo.handlers.gevent

import json
import time
import disc_consts

from gevent.coros import BoundedSemaphore


class DiscoveryZkClient(object):

    def __init__(self, discServer, zk_srv_ip='127.0.0.1',
                 zk_srv_port='2181', reset_config=False):
        self._reset_config = reset_config
        self._service_id_to_type = {}
        self._ds = discServer
        self._zk_sem = BoundedSemaphore(1)

        self._zk = kazoo.client.KazooClient(
            hosts='%s:%s' % (zk_srv_ip, zk_srv_port), timeout=120,
            handler=kazoo.handlers.gevent.SequentialGeventHandler())

        # connect
        while True:
            try:
                self._zk.start()
                break
            except gevent.event.Timeout as e:
                self.syslog(
                    'Failed to connect with Zookeeper -will retry in a second')
                gevent.sleep(1)
        self.syslog('Connected to ZooKeeper!')

        if reset_config:
            self._zk.delete("/services", recursive=True)
            self._zk.delete("/clients", recursive=True)
            self._zk.delete("/election", recursive=True)

        # create default paths
        self.create_node("/services")
        self.create_node("/clients")
        self.create_node("/election")

        self._debug = {
            'subscription_expires': 0,
            'oos_delete': 0,
            'db_excepts': 0,
        }
    # end __init__

    def start_background_tasks(self):
        # spawn loop to expire subscriptions
        gevent.Greenlet.spawn(self.inuse_loop)

        # spawn loop to expire services
        gevent.Greenlet.spawn(self.service_oos_loop)
    # end

    def syslog(self, log_msg):
        self._ds.syslog(log_msg)
    # end

    def get_debug_stats(self):
        return self._debug
    # end

    def sem_acquire(self):
        self._zk_sem.acquire()
    # end

    def sem_release(self):
        self._zk_sem.release()
    # end

    def create_node(self, path, value='', makepath=False):
        try:
            self._zk.set(path, value)
        except kazoo.exceptions.NoNodeException:
            self._zk.create(path, value, makepath=makepath)
            self.syslog('create %s' % (path))
    # end create_node

    def service_entries(self):
        service_types = self._zk.get_children('/services')
        for service_type in service_types:
            services = self._zk.get_children('/services/%s' % (service_type))
            for service_id in services:
                data, stat = self._zk.get(
                    '/services/%s/%s' % (service_type, service_id))
                entry = json.loads(data)
                yield(entry)

    def subscriber_entries(self):
        service_types = self._zk.get_children('/clients')
        for service_type in service_types:
            subscribers = self._zk.get_children('/clients/%s' % (service_type))
            for client_id in subscribers:
                data, stat = self._zk.get(
                    '/clients/%s/%s' % (service_type, client_id))
                cl_entry = json.loads(data)
                yield((client_id, service_type))
    # end

    def update_service(self, service_type, service_id, data):
        path = '/services/%s/%s' % (service_type, service_id)
        self.create_node(path, value=json.dumps(data), makepath=True)
    # end
<commit_msg>reduce zookeeper exceptions (and logs)<commit_after>#
# Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
#

import re
import gevent
import kazoo.client
import kazoo.exceptions
import kazoo.handlers.gevent

import json
import time
import disc_consts

from gevent.coros import BoundedSemaphore


class DiscoveryZkClient(object):

    def __init__(self, discServer, zk_srv_ip='127.0.0.1',
                 zk_srv_port='2181', reset_config=False):
        self._reset_config = reset_config
        self._service_id_to_type = {}
        self._ds = discServer
        self._zk_sem = BoundedSemaphore(1)

        self._zk = kazoo.client.KazooClient(
            hosts='%s:%s' % (zk_srv_ip, zk_srv_port), timeout=120,
            handler=kazoo.handlers.gevent.SequentialGeventHandler())

        # connect
        while True:
            try:
                self._zk.start()
                break
            except gevent.event.Timeout as e:
                self.syslog(
                    'Failed to connect with Zookeeper -will retry in a second')
                gevent.sleep(1)
        self.syslog('Connected to ZooKeeper!')

        if reset_config:
            self._zk.delete("/services", recursive=True)
            self._zk.delete("/clients", recursive=True)
            self._zk.delete("/election", recursive=True)

        # create default paths
        self.create_node("/services")
        self.create_node("/clients")
        self.create_node("/election")

        self._debug = {
            'subscription_expires': 0,
            'oos_delete': 0,
            'db_excepts': 0,
        }
    # end __init__

    def start_background_tasks(self):
        # spawn loop to expire subscriptions
        gevent.Greenlet.spawn(self.inuse_loop)

        # spawn loop to expire services
        gevent.Greenlet.spawn(self.service_oos_loop)
    # end

    def syslog(self, log_msg):
        self._ds.syslog(log_msg)
    # end

    def get_debug_stats(self):
        return self._debug
    # end

    def sem_acquire(self):
        self._zk_sem.acquire()
    # end

    def sem_release(self):
        self._zk_sem.release()
    # end

    def create_node(self, path, value='', makepath=False):
        if self._zk.exists(path):
            self._zk.set(path, value)
        else:
            self._zk.create(path, value, makepath=makepath)
            self.syslog('create %s' % (path))
    # end create_node

    def service_entries(self):
        service_types = self._zk.get_children('/services')
        for service_type in service_types:
            services = self._zk.get_children('/services/%s' % (service_type))
            for service_id in services:
                data, stat = self._zk.get(
                    '/services/%s/%s' % (service_type, service_id))
                entry = json.loads(data)
                yield(entry)

    def subscriber_entries(self):
        service_types = self._zk.get_children('/clients')
        for service_type in service_types:
            subscribers = self._zk.get_children('/clients/%s' % (service_type))
            for client_id in subscribers:
                data, stat = self._zk.get(
                    '/clients/%s/%s' % (service_type, client_id))
                cl_entry = json.loads(data)
                yield((client_id, service_type))
    # end

    def update_service(self, service_type, service_id, data):
        path = '/services/%s/%s' % (service_type, service_id)
        self.create_node(path, value=json.dumps(data), makepath=True)
    # end

{'fraction_non_alphanumeric': 0.10699588477366255, 'fraction_numerical': 0.006243791684404711, 'mean_word_length': 2.098021978021978, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '780188', 'n_tokens_mistral': 2220, 'n_tokens_neox': 2106, 'n_words': 432}