We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[core] network_check active threads count causing check aborts
There's a [check on network_checks.py:120](https://github.com/DataDog/dd-agent/blob/6df4554465338b22441f5bba04913538fe7ef1a2/checks/network_checks.py#L120-L121) that looks to see if there are too many active threads.

```Python
        if threading.activeCount() > 5 * self.pool_size + 5: # On Windows the agent runs on multiple threads so we need to have an offset of 5 in case the pool_size is 1
            raise Exception("Thread number (%s) is exploding. Skipping this check" % threading.activeCount())
```

This works fine as long as you only have one network_check running on a box or otherwise have network checks configured with similar values for `pool_size`. This breaks when you have one or more large pools alongside smaller ones. This is easy situation to get into if you have a large number of instances in one network_check as it sets the pool size to the number of instances.

All of the big pools instance check just fine, but the smaller network_checks never run. We have a custom snmpwalk based check with a VERY large number of instances that runs alongside a small set. As soon as we turned on the large set we lost the small one.

**Additional environment details (Operating System, Cloud provider, etc):**


**Steps to reproduce the issue:**
1. Configure a `NetworkCheck` derived check with a large number of instances each of which will be active for a bit of time
2. Configure another `NetworkCheck` derived check with a small number of instances, e.g. 1

**Describe the results you received:**

```
2016-12-13 12:33:25 PST | ERROR | dd.collector | checks.snmpwalk(__init__.py:788) | Check 'snmpwalk' instance #0 failed
Traceback (most recent call last):
  File "/opt/datadog-agent/agent/checks/__init__.py", line 771, in run
    self.check(copy.deepcopy(instance))
  File "/opt/datadog-agent/agent/checks/network_checks.py", line 122, in check
    raise Exception("Thread number (%s) is exploding. Skipping this check" % threading.activeCount())
Exception: Thread number (32) is exploding. Skipping this check
```

**Describe the results you expected:**

The active thread count should be looked at globally before kicking off the whole check process and take into account the sum of all the pool_sizes rather than compare the global active_threads to a per-instance pool_size.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `checks/network_checks.py`
Content:
```
1 # (C) Datadog, Inc. 2010-2016
2 # All rights reserved
3 # Licensed under Simplified BSD License (see LICENSE)
4 
5 # stdlib
6 from collections import defaultdict
7 from Queue import Empty, Queue
8 import threading
9 import time
10 
11 # project
12 from checks import AgentCheck
13 from checks.libs.thread_pool import Pool
14 from config import _is_affirmative
15 
16 TIMEOUT = 180
17 DEFAULT_SIZE_POOL = 6
18 MAX_LOOP_ITERATIONS = 1000
19 FAILURE = "FAILURE"
20 
21 
22 class Status:
23     DOWN = "DOWN"
24     WARNING = "WARNING"
25     CRITICAL = "CRITICAL"
26     UP = "UP"
27 
28 
29 class EventType:
30     DOWN = "servicecheck.state_change.down"
31     UP = "servicecheck.state_change.up"
32 
33 
34 class NetworkCheck(AgentCheck):
35     SOURCE_TYPE_NAME = 'servicecheck'
36     SERVICE_CHECK_PREFIX = 'network_check'
37 
38     STATUS_TO_SERVICE_CHECK = {
39         Status.UP : AgentCheck.OK,
40         Status.WARNING : AgentCheck.WARNING,
41         Status.CRITICAL : AgentCheck.CRITICAL,
42         Status.DOWN : AgentCheck.CRITICAL,
43     }
44 
45     """
46     Services checks inherits from this class.
47     This class should never be directly instanciated.
48 
49     Work flow:
50         The main agent loop will call the check function for each instance for
51         each iteration of the loop.
52         The check method will make an asynchronous call to the _process method in
53         one of the thread initiated in the thread pool created in this class constructor.
54         The _process method will call the _check method of the inherited class
55         which will perform the actual check.
56 
57         The _check method must return a tuple which first element is either
58             Status.UP or Status.DOWN.
59             The second element is a short error message that will be displayed
60             when the service turns down.
61 
62     """
63 
64     def __init__(self, name, init_config, agentConfig, instances):
65         AgentCheck.__init__(self, name, init_config, agentConfig, instances)
66 
67         # A dictionary to keep track of service statuses
68         self.statuses = {}
69         self.notified = {}
70         self.nb_failures = 0
71         self.pool_started = False
72 
73         # Make sure every instance has a name that we use as a unique key
74         # to keep track of statuses
75         names = []
76         for inst in instances:
77             inst_name = inst.get('name', None)
78             if not inst_name:
79                 raise Exception("All instances should have a 'name' parameter,"
80                                 " error on instance: {0}".format(inst))
81             if inst_name in names:
82                 raise Exception("Duplicate names for instances with name {0}"
83                                 .format(inst_name))
84             names.append(inst_name)
85 
86     def stop(self):
87         self.stop_pool()
88         self.pool_started = False
89 
90     def start_pool(self):
91         # The pool size should be the minimum between the number of instances
92         # and the DEFAULT_SIZE_POOL. It can also be overridden by the 'threads_count'
93         # parameter in the init_config of the check
94         self.log.info("Starting Thread Pool")
95         default_size = min(self.instance_count(), DEFAULT_SIZE_POOL)
96         self.pool_size = int(self.init_config.get('threads_count', default_size))
97 
98         self.pool = Pool(self.pool_size)
99 
100         self.resultsq = Queue()
101         self.jobs_status = {}
102         self.jobs_results = {}
103         self.pool_started = True
104 
105     def stop_pool(self):
106         self.log.info("Stopping Thread Pool")
107         if self.pool_started:
108             self.pool.terminate()
109             self.pool.join()
110             self.jobs_status.clear()
111             assert self.pool.get_nworkers() == 0
112 
113     def restart_pool(self):
114         self.stop_pool()
115         self.start_pool()
116 
117     def check(self, instance):
118         if not self.pool_started:
119             self.start_pool()
120         if threading.activeCount() > 5 * self.pool_size + 5: # On Windows the agent runs on multiple threads so we need to have an offset of 5 in case the pool_size is 1
121             raise Exception("Thread number (%s) is exploding. Skipping this check" % threading.activeCount())
122         self._process_results()
123         self._clean()
124         name = instance.get('name', None)
125         if name is None:
126             self.log.error('Each service check must have a name')
127             return
128 
129         if name not in self.jobs_status:
130             # A given instance should be processed one at a time
131             self.jobs_status[name] = time.time()
132             self.jobs_results[name] = self.pool.apply_async(self._process, args=(instance,))
133         else:
134             self.log.error("Instance: %s skipped because it's already running." % name)
135 
136     def _process(self, instance):
137         try:
138             statuses = self._check(instance)
139 
140             if isinstance(statuses, tuple):
141                 # Assume the check only returns one service check
142                 status, msg = statuses
143                 self.resultsq.put((status, msg, None, instance))
144 
145             elif isinstance(statuses, list):
146                 for status in statuses:
147                     sc_name, status, msg = status
148                     self.resultsq.put((status, msg, sc_name, instance))
149 
150         except Exception:
151             self.log.exception(
152                 u"Failed to process instance '%s'.", instance.get('Name', u"")
153             )
154             result = (FAILURE, FAILURE, FAILURE, instance)
155             self.resultsq.put(result)
156 
157     def _process_results(self):
158         for i in xrange(MAX_LOOP_ITERATIONS):
159             try:
160                 # We want to fetch the result in a non blocking way
161                 status, msg, sc_name, instance = self.resultsq.get_nowait()
162             except Empty:
163                 break
164 
165             instance_name = instance['name']
166             if status == FAILURE:
167                 self.nb_failures += 1
168                 if self.nb_failures >= self.pool_size - 1:
169                     self.nb_failures = 0
170                     self.restart_pool()
171 
172                 # clean failed job
173                 self._clean_job(instance_name)
174                 continue
175 
176             self.report_as_service_check(sc_name, status, instance, msg)
177 
178             # FIXME: 5.3, this has been deprecated before, get rid of events
179             # Don't create any event to avoid duplicates with server side
180             # service_checks
181             skip_event = _is_affirmative(instance.get('skip_event', False))
182             if not skip_event:
183                 self.warning("Using events for service checks is deprecated in favor of monitors and will be removed in future versions of the Datadog Agent.")
184                 event = None
185 
186                 if instance_name not in self.statuses:
187                     self.statuses[instance_name] = defaultdict(list)
188 
189                 self.statuses[instance_name][sc_name].append(status)
190 
191                 window = int(instance.get('window', 1))
192 
193                 if window > 256:
194                     self.log.warning("Maximum window size (256) exceeded, defaulting it to 256")
195                     window = 256
196 
197                 threshold = instance.get('threshold', 1)
198 
199                 if len(self.statuses[instance_name][sc_name]) > window:
200                     self.statuses[instance_name][sc_name].pop(0)
201 
202                 nb_failures = self.statuses[instance_name][sc_name].count(Status.DOWN)
203 
204                 if nb_failures >= threshold:
205                     if self.notified.get((instance_name, sc_name), Status.UP) != Status.DOWN:
206                         event = self._create_status_event(sc_name, status, msg, instance)
207                         self.notified[(instance_name, sc_name)] = Status.DOWN
208                 else:
209                     if self.notified.get((instance_name, sc_name), Status.UP) != Status.UP:
210                         event = self._create_status_event(sc_name, status, msg, instance)
211                         self.notified[(instance_name, sc_name)] = Status.UP
212 
213                 if event is not None:
214                     self.events.append(event)
215 
216             self._clean_job(instance_name)
217 
218     def _clean_job(self, instance_name):
219         # The job is finished here, this instance can be re processed
220         if instance_name in self.jobs_status:
221             self.log.debug("Instance: %s cleaned from jobs status." % instance_name)
222             del self.jobs_status[instance_name]
223 
224         # if an exception happened, log it
225         if instance_name in self.jobs_results:
226             self.log.debug("Instance: %s cleaned from jobs results." % instance_name)
227             ret = self.jobs_results[instance_name].get()
228             if isinstance(ret, Exception):
229                 self.log.exception("Exception in worker thread: {0}".format(ret))
230             del self.jobs_results[instance_name]
231 
232 
233     def _check(self, instance):
234         """This function should be implemented by inherited classes"""
235         raise NotImplementedError
236 
237 
238     def _clean(self):
239         now = time.time()
240         for name, start_time in self.jobs_status.iteritems():
241             if now - start_time > TIMEOUT:
242                 self.log.critical("Restarting Pool. One check is stuck: %s" % name)
243                 self.restart_pool()
244                 break
245
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/checks/network_checks.py b/checks/network_checks.py
--- a/checks/network_checks.py
+++ b/checks/network_checks.py
@@ -18,7 +18,6 @@
 MAX_LOOP_ITERATIONS = 1000
 FAILURE = "FAILURE"
 
-
 class Status:
     DOWN = "DOWN"
     WARNING = "WARNING"
@@ -34,6 +33,7 @@
 class NetworkCheck(AgentCheck):
     SOURCE_TYPE_NAME = 'servicecheck'
     SERVICE_CHECK_PREFIX = 'network_check'
+    _global_current_pool_size = 0
 
     STATUS_TO_SERVICE_CHECK = {
         Status.UP : AgentCheck.OK,
@@ -68,6 +68,7 @@
         self.statuses = {}
         self.notified = {}
         self.nb_failures = 0
+        self.pool_size = 0
         self.pool_started = False
 
         # Make sure every instance has a name that we use as a unique key
@@ -95,6 +96,9 @@
         default_size = min(self.instance_count(), DEFAULT_SIZE_POOL)
         self.pool_size = int(self.init_config.get('threads_count', default_size))
 
+        # To keep track on the total number of threads we should have running
+        NetworkCheck._global_current_pool_size += self.pool_size
+
         self.pool = Pool(self.pool_size)
 
         self.resultsq = Queue()
@@ -104,6 +108,10 @@
 
     def stop_pool(self):
         self.log.info("Stopping Thread Pool")
+
+        # To keep track on the total number of threads we should have running
+        NetworkCheck._global_current_pool_size -= self.pool_size
+
         if self.pool_started:
             self.pool.terminate()
             self.pool.join()
@@ -117,7 +125,8 @@
     def check(self, instance):
         if not self.pool_started:
             self.start_pool()
-        if threading.activeCount() > 5 * self.pool_size + 5: # On Windows the agent runs on multiple threads so we need to have an offset of 5 in case the pool_size is 1
+        if threading.activeCount() > 5 * NetworkCheck._global_current_pool_size + 6:
+            # On Windows the agent runs on multiple threads because of WMI so we need an offset of 6
             raise Exception("Thread number (%s) is exploding. Skipping this check" % threading.activeCount())
         self._process_results()
         self._clean()

{"golden_diff": "diff --git a/checks/network_checks.py b/checks/network_checks.py\n--- a/checks/network_checks.py\n+++ b/checks/network_checks.py\n@@ -18,7 +18,6 @@\n MAX_LOOP_ITERATIONS = 1000\n FAILURE = \"FAILURE\"\n \n-\n class Status:\n     DOWN = \"DOWN\"\n     WARNING = \"WARNING\"\n@@ -34,6 +33,7 @@\n class NetworkCheck(AgentCheck):\n     SOURCE_TYPE_NAME = 'servicecheck'\n     SERVICE_CHECK_PREFIX = 'network_check'\n+    _global_current_pool_size = 0\n \n     STATUS_TO_SERVICE_CHECK = {\n         Status.UP : AgentCheck.OK,\n@@ -68,6 +68,7 @@\n         self.statuses = {}\n         self.notified = {}\n         self.nb_failures = 0\n+        self.pool_size = 0\n         self.pool_started = False\n \n         # Make sure every instance has a name that we use as a unique key\n@@ -95,6 +96,9 @@\n         default_size = min(self.instance_count(), DEFAULT_SIZE_POOL)\n         self.pool_size = int(self.init_config.get('threads_count', default_size))\n \n+        # To keep track on the total number of threads we should have running\n+        NetworkCheck._global_current_pool_size += self.pool_size\n+\n         self.pool = Pool(self.pool_size)\n \n         self.resultsq = Queue()\n@@ -104,6 +108,10 @@\n \n     def stop_pool(self):\n         self.log.info(\"Stopping Thread Pool\")\n+\n+        # To keep track on the total number of threads we should have running\n+        NetworkCheck._global_current_pool_size -= self.pool_size\n+\n         if self.pool_started:\n             self.pool.terminate()\n             self.pool.join()\n@@ -117,7 +125,8 @@\n     def check(self, instance):\n         if not self.pool_started:\n             self.start_pool()\n-        if threading.activeCount() > 5 * self.pool_size + 5: # On Windows the agent runs on multiple threads so we need to have an offset of 5 in case the pool_size is 1\n+        if threading.activeCount() > 5 * NetworkCheck._global_current_pool_size + 6:\n+            # On Windows the agent runs on multiple threads because of WMI so we need an offset of 6\n             raise Exception(\"Thread number (%s) is exploding. Skipping this check\" % threading.activeCount())\n         self._process_results()\n         self._clean()\n", "issue": "[core] network_check active threads count causing check aborts\nThere's a [check on network_checks.py:120](https://github.com/DataDog/dd-agent/blob/6df4554465338b22441f5bba04913538fe7ef1a2/checks/network_checks.py#L120-L121) that looks to see if there are too many active threads.\r\n\r\n```Python\r\n        if threading.activeCount() > 5 * self.pool_size + 5: # On Windows the agent runs on multiple threads so we need to have an offset of 5 in case the pool_size is 1\r\n            raise Exception(\"Thread number (%s) is exploding. Skipping this check\" % threading.activeCount())\r\n```\r\n\r\nThis works fine as long as you only have one network_check running on a box or otherwise have network checks configured with similar values for `pool_size`. This breaks when you have one or more large pools alongside smaller ones. This is easy situation to get into if you have a large number of instances in one network_check as it sets the pool size to the number of instances.\r\n\r\nAll of the big pools instance check just fine, but the smaller network_checks never run. We have a custom snmpwalk based check with a VERY large number of instances that runs alongside a small set. As soon as we turned on the large set we lost the small one.\r\n\r\n**Additional environment details (Operating System, Cloud provider, etc):**\r\n\r\n\r\n**Steps to reproduce the issue:**\r\n1. Configure a `NetworkCheck` derived check with a large number of instances each of which will be active for a bit of time\r\n2. Configure another `NetworkCheck` derived check with a small number of instances, e.g. 1\r\n\r\n**Describe the results you received:**\r\n\r\n```\r\n2016-12-13 12:33:25 PST | ERROR | dd.collector | checks.snmpwalk(__init__.py:788) | Check 'snmpwalk' instance #0 failed\r\nTraceback (most recent call last):\r\n  File \"/opt/datadog-agent/agent/checks/__init__.py\", line 771, in run\r\n    self.check(copy.deepcopy(instance))\r\n  File \"/opt/datadog-agent/agent/checks/network_checks.py\", line 122, in check\r\n    raise Exception(\"Thread number (%s) is exploding. Skipping this check\" % threading.activeCount())\r\nException: Thread number (32) is exploding. Skipping this check\r\n```\r\n\r\n**Describe the results you expected:**\r\n\r\nThe active thread count should be looked at globally before kicking off the whole check process and take into account the sum of all the pool_sizes rather than compare the global active_threads to a per-instance pool_size.\n", "before_files": [{"content": "# (C) Datadog, Inc. 2010-2016\n# All rights reserved\n# Licensed under Simplified BSD License (see LICENSE)\n\n# stdlib\nfrom collections import defaultdict\nfrom Queue import Empty, Queue\nimport threading\nimport time\n\n# project\nfrom checks import AgentCheck\nfrom checks.libs.thread_pool import Pool\nfrom config import _is_affirmative\n\nTIMEOUT = 180\nDEFAULT_SIZE_POOL = 6\nMAX_LOOP_ITERATIONS = 1000\nFAILURE = \"FAILURE\"\n\n\nclass Status:\n    DOWN = \"DOWN\"\n    WARNING = \"WARNING\"\n    CRITICAL = \"CRITICAL\"\n    UP = \"UP\"\n\n\nclass EventType:\n    DOWN = \"servicecheck.state_change.down\"\n    UP = \"servicecheck.state_change.up\"\n\n\nclass NetworkCheck(AgentCheck):\n    SOURCE_TYPE_NAME = 'servicecheck'\n    SERVICE_CHECK_PREFIX = 'network_check'\n\n    STATUS_TO_SERVICE_CHECK = {\n        Status.UP : AgentCheck.OK,\n        Status.WARNING : AgentCheck.WARNING,\n        Status.CRITICAL : AgentCheck.CRITICAL,\n        Status.DOWN : AgentCheck.CRITICAL,\n    }\n\n    \"\"\"\n    Services checks inherits from this class.\n    This class should never be directly instanciated.\n\n    Work flow:\n        The main agent loop will call the check function for each instance for\n        each iteration of the loop.\n        The check method will make an asynchronous call to the _process method in\n        one of the thread initiated in the thread pool created in this class constructor.\n        The _process method will call the _check method of the inherited class\n        which will perform the actual check.\n\n        The _check method must return a tuple which first element is either\n            Status.UP or Status.DOWN.\n            The second element is a short error message that will be displayed\n            when the service turns down.\n\n    \"\"\"\n\n    def __init__(self, name, init_config, agentConfig, instances):\n        AgentCheck.__init__(self, name, init_config, agentConfig, instances)\n\n        # A dictionary to keep track of service statuses\n        self.statuses = {}\n        self.notified = {}\n        self.nb_failures = 0\n        self.pool_started = False\n\n        # Make sure every instance has a name that we use as a unique key\n        # to keep track of statuses\n        names = []\n        for inst in instances:\n            inst_name = inst.get('name', None)\n            if not inst_name:\n                raise Exception(\"All instances should have a 'name' parameter,\"\n                                \" error on instance: {0}\".format(inst))\n            if inst_name in names:\n                raise Exception(\"Duplicate names for instances with name {0}\"\n                                .format(inst_name))\n            names.append(inst_name)\n\n    def stop(self):\n        self.stop_pool()\n        self.pool_started = False\n\n    def start_pool(self):\n        # The pool size should be the minimum between the number of instances\n        # and the DEFAULT_SIZE_POOL. It can also be overridden by the 'threads_count'\n        # parameter in the init_config of the check\n        self.log.info(\"Starting Thread Pool\")\n        default_size = min(self.instance_count(), DEFAULT_SIZE_POOL)\n        self.pool_size = int(self.init_config.get('threads_count', default_size))\n\n        self.pool = Pool(self.pool_size)\n\n        self.resultsq = Queue()\n        self.jobs_status = {}\n        self.jobs_results = {}\n        self.pool_started = True\n\n    def stop_pool(self):\n        self.log.info(\"Stopping Thread Pool\")\n        if self.pool_started:\n            self.pool.terminate()\n            self.pool.join()\n            self.jobs_status.clear()\n            assert self.pool.get_nworkers() == 0\n\n    def restart_pool(self):\n        self.stop_pool()\n        self.start_pool()\n\n    def check(self, instance):\n        if not self.pool_started:\n            self.start_pool()\n        if threading.activeCount() > 5 * self.pool_size + 5: # On Windows the agent runs on multiple threads so we need to have an offset of 5 in case the pool_size is 1\n            raise Exception(\"Thread number (%s) is exploding. Skipping this check\" % threading.activeCount())\n        self._process_results()\n        self._clean()\n        name = instance.get('name', None)\n        if name is None:\n            self.log.error('Each service check must have a name')\n            return\n\n        if name not in self.jobs_status:\n            # A given instance should be processed one at a time\n            self.jobs_status[name] = time.time()\n            self.jobs_results[name] = self.pool.apply_async(self._process, args=(instance,))\n        else:\n            self.log.error(\"Instance: %s skipped because it's already running.\" % name)\n\n    def _process(self, instance):\n        try:\n            statuses = self._check(instance)\n\n            if isinstance(statuses, tuple):\n                # Assume the check only returns one service check\n                status, msg = statuses\n                self.resultsq.put((status, msg, None, instance))\n\n            elif isinstance(statuses, list):\n                for status in statuses:\n                    sc_name, status, msg = status\n                    self.resultsq.put((status, msg, sc_name, instance))\n\n        except Exception:\n            self.log.exception(\n                u\"Failed to process instance '%s'.\", instance.get('Name', u\"\")\n            )\n            result = (FAILURE, FAILURE, FAILURE, instance)\n            self.resultsq.put(result)\n\n    def _process_results(self):\n        for i in xrange(MAX_LOOP_ITERATIONS):\n            try:\n                # We want to fetch the result in a non blocking way\n                status, msg, sc_name, instance = self.resultsq.get_nowait()\n            except Empty:\n                break\n\n            instance_name = instance['name']\n            if status == FAILURE:\n                self.nb_failures += 1\n                if self.nb_failures >= self.pool_size - 1:\n                    self.nb_failures = 0\n                    self.restart_pool()\n\n                # clean failed job\n                self._clean_job(instance_name)\n                continue\n\n            self.report_as_service_check(sc_name, status, instance, msg)\n\n            # FIXME: 5.3, this has been deprecated before, get rid of events\n            # Don't create any event to avoid duplicates with server side\n            # service_checks\n            skip_event = _is_affirmative(instance.get('skip_event', False))\n            if not skip_event:\n                self.warning(\"Using events for service checks is deprecated in favor of monitors and will be removed in future versions of the Datadog Agent.\")\n                event = None\n\n                if instance_name not in self.statuses:\n                    self.statuses[instance_name] = defaultdict(list)\n\n                self.statuses[instance_name][sc_name].append(status)\n\n                window = int(instance.get('window', 1))\n\n                if window > 256:\n                    self.log.warning(\"Maximum window size (256) exceeded, defaulting it to 256\")\n                    window = 256\n\n                threshold = instance.get('threshold', 1)\n\n                if len(self.statuses[instance_name][sc_name]) > window:\n                    self.statuses[instance_name][sc_name].pop(0)\n\n                nb_failures = self.statuses[instance_name][sc_name].count(Status.DOWN)\n\n                if nb_failures >= threshold:\n                    if self.notified.get((instance_name, sc_name), Status.UP) != Status.DOWN:\n                        event = self._create_status_event(sc_name, status, msg, instance)\n                        self.notified[(instance_name, sc_name)] = Status.DOWN\n                else:\n                    if self.notified.get((instance_name, sc_name), Status.UP) != Status.UP:\n                        event = self._create_status_event(sc_name, status, msg, instance)\n                        self.notified[(instance_name, sc_name)] = Status.UP\n\n                if event is not None:\n                    self.events.append(event)\n\n            self._clean_job(instance_name)\n\n    def _clean_job(self, instance_name):\n        # The job is finished here, this instance can be re processed\n        if instance_name in self.jobs_status:\n            self.log.debug(\"Instance: %s cleaned from jobs status.\" % instance_name)\n            del self.jobs_status[instance_name]\n\n        # if an exception happened, log it\n        if instance_name in self.jobs_results:\n            self.log.debug(\"Instance: %s cleaned from jobs results.\" % instance_name)\n            ret = self.jobs_results[instance_name].get()\n            if isinstance(ret, Exception):\n                self.log.exception(\"Exception in worker thread: {0}\".format(ret))\n            del self.jobs_results[instance_name]\n\n\n    def _check(self, instance):\n        \"\"\"This function should be implemented by inherited classes\"\"\"\n        raise NotImplementedError\n\n\n    def _clean(self):\n        now = time.time()\n        for name, start_time in self.jobs_status.iteritems():\n            if now - start_time > TIMEOUT:\n                self.log.critical(\"Restarting Pool. One check is stuck: %s\" % name)\n                self.restart_pool()\n                break\n", "path": "checks/network_checks.py"}], "after_files": [{"content": "# (C) Datadog, Inc. 2010-2016\n# All rights reserved\n# Licensed under Simplified BSD License (see LICENSE)\n\n# stdlib\nfrom collections import defaultdict\nfrom Queue import Empty, Queue\nimport threading\nimport time\n\n# project\nfrom checks import AgentCheck\nfrom checks.libs.thread_pool import Pool\nfrom config import _is_affirmative\n\nTIMEOUT = 180\nDEFAULT_SIZE_POOL = 6\nMAX_LOOP_ITERATIONS = 1000\nFAILURE = \"FAILURE\"\n\nclass Status:\n    DOWN = \"DOWN\"\n    WARNING = \"WARNING\"\n    CRITICAL = \"CRITICAL\"\n    UP = \"UP\"\n\n\nclass EventType:\n    DOWN = \"servicecheck.state_change.down\"\n    UP = \"servicecheck.state_change.up\"\n\n\nclass NetworkCheck(AgentCheck):\n    SOURCE_TYPE_NAME = 'servicecheck'\n    SERVICE_CHECK_PREFIX = 'network_check'\n    _global_current_pool_size = 0\n\n    STATUS_TO_SERVICE_CHECK = {\n        Status.UP : AgentCheck.OK,\n        Status.WARNING : AgentCheck.WARNING,\n        Status.CRITICAL : AgentCheck.CRITICAL,\n        Status.DOWN : AgentCheck.CRITICAL,\n    }\n\n    \"\"\"\n    Services checks inherits from this class.\n    This class should never be directly instanciated.\n\n    Work flow:\n        The main agent loop will call the check function for each instance for\n        each iteration of the loop.\n        The check method will make an asynchronous call to the _process method in\n        one of the thread initiated in the thread pool created in this class constructor.\n        The _process method will call the _check method of the inherited class\n        which will perform the actual check.\n\n        The _check method must return a tuple which first element is either\n            Status.UP or Status.DOWN.\n            The second element is a short error message that will be displayed\n            when the service turns down.\n\n    \"\"\"\n\n    def __init__(self, name, init_config, agentConfig, instances):\n        AgentCheck.__init__(self, name, init_config, agentConfig, instances)\n\n        # A dictionary to keep track of service statuses\n        self.statuses = {}\n        self.notified = {}\n        self.nb_failures = 0\n        self.pool_size = 0\n        self.pool_started = False\n\n        # Make sure every instance has a name that we use as a unique key\n        # to keep track of statuses\n        names = []\n        for inst in instances:\n            inst_name = inst.get('name', None)\n            if not inst_name:\n                raise Exception(\"All instances should have a 'name' parameter,\"\n                                \" error on instance: {0}\".format(inst))\n            if inst_name in names:\n                raise Exception(\"Duplicate names for instances with name {0}\"\n                                .format(inst_name))\n            names.append(inst_name)\n\n    def stop(self):\n        self.stop_pool()\n        self.pool_started = False\n\n    def start_pool(self):\n        # The pool size should be the minimum between the number of instances\n        # and the DEFAULT_SIZE_POOL. It can also be overridden by the 'threads_count'\n        # parameter in the init_config of the check\n        self.log.info(\"Starting Thread Pool\")\n        default_size = min(self.instance_count(), DEFAULT_SIZE_POOL)\n        self.pool_size = int(self.init_config.get('threads_count', default_size))\n\n        # To keep track on the total number of threads we should have running\n        NetworkCheck._global_current_pool_size += self.pool_size\n\n        self.pool = Pool(self.pool_size)\n\n        self.resultsq = Queue()\n        self.jobs_status = {}\n        self.jobs_results = {}\n        self.pool_started = True\n\n    def stop_pool(self):\n        self.log.info(\"Stopping Thread Pool\")\n\n        # To keep track on the total number of threads we should have running\n        NetworkCheck._global_current_pool_size -= self.pool_size\n\n        if self.pool_started:\n            self.pool.terminate()\n            self.pool.join()\n            self.jobs_status.clear()\n            assert self.pool.get_nworkers() == 0\n\n    def restart_pool(self):\n        self.stop_pool()\n        self.start_pool()\n\n    def check(self, instance):\n        if not self.pool_started:\n            self.start_pool()\n        if threading.activeCount() > 5 * NetworkCheck._global_current_pool_size + 6:\n            # On Windows the agent runs on multiple threads because of WMI so we need an offset of 6\n            raise Exception(\"Thread number (%s) is exploding. Skipping this check\" % threading.activeCount())\n        self._process_results()\n        self._clean()\n        name = instance.get('name', None)\n        if name is None:\n            self.log.error('Each service check must have a name')\n            return\n\n        if name not in self.jobs_status:\n            # A given instance should be processed one at a time\n            self.jobs_status[name] = time.time()\n            self.jobs_results[name] = self.pool.apply_async(self._process, args=(instance,))\n        else:\n            self.log.error(\"Instance: %s skipped because it's already running.\" % name)\n\n    def _process(self, instance):\n        try:\n            statuses = self._check(instance)\n\n            if isinstance(statuses, tuple):\n                # Assume the check only returns one service check\n                status, msg = statuses\n                self.resultsq.put((status, msg, None, instance))\n\n            elif isinstance(statuses, list):\n                for status in statuses:\n                    sc_name, status, msg = status\n                    self.resultsq.put((status, msg, sc_name, instance))\n\n        except Exception:\n            self.log.exception(\n                u\"Failed to process instance '%s'.\", instance.get('Name', u\"\")\n            )\n            result = (FAILURE, FAILURE, FAILURE, instance)\n            self.resultsq.put(result)\n\n    def _process_results(self):\n        for i in xrange(MAX_LOOP_ITERATIONS):\n            try:\n                # We want to fetch the result in a non blocking way\n                status, msg, sc_name, instance = self.resultsq.get_nowait()\n            except Empty:\n                break\n\n            instance_name = instance['name']\n            if status == FAILURE:\n                self.nb_failures += 1\n                if self.nb_failures >= self.pool_size - 1:\n                    self.nb_failures = 0\n                    self.restart_pool()\n\n                # clean failed job\n                self._clean_job(instance_name)\n                continue\n\n            self.report_as_service_check(sc_name, status, instance, msg)\n\n            # FIXME: 5.3, this has been deprecated before, get rid of events\n            # Don't create any event to avoid duplicates with server side\n            # service_checks\n            skip_event = _is_affirmative(instance.get('skip_event', False))\n            if not skip_event:\n                self.warning(\"Using events for service checks is deprecated in favor of monitors and will be removed in future versions of the Datadog Agent.\")\n                event = None\n\n                if instance_name not in self.statuses:\n                    self.statuses[instance_name] = defaultdict(list)\n\n                self.statuses[instance_name][sc_name].append(status)\n\n                window = int(instance.get('window', 1))\n\n                if window > 256:\n                    self.log.warning(\"Maximum window size (256) exceeded, defaulting it to 256\")\n                    window = 256\n\n                threshold = instance.get('threshold', 1)\n\n                if len(self.statuses[instance_name][sc_name]) > window:\n                    self.statuses[instance_name][sc_name].pop(0)\n\n                nb_failures = self.statuses[instance_name][sc_name].count(Status.DOWN)\n\n                if nb_failures >= threshold:\n                    if self.notified.get((instance_name, sc_name), Status.UP) != Status.DOWN:\n                        event = self._create_status_event(sc_name, status, msg, instance)\n                        self.notified[(instance_name, sc_name)] = Status.DOWN\n                else:\n                    if self.notified.get((instance_name, sc_name), Status.UP) != Status.UP:\n                        event = self._create_status_event(sc_name, status, msg, instance)\n                        self.notified[(instance_name, sc_name)] = Status.UP\n\n                if event is not None:\n                    self.events.append(event)\n\n            self._clean_job(instance_name)\n\n    def _clean_job(self, instance_name):\n        # The job is finished here, this instance can be re processed\n        if instance_name in self.jobs_status:\n            self.log.debug(\"Instance: %s cleaned from jobs status.\" % instance_name)\n            del self.jobs_status[instance_name]\n\n        # if an exception happened, log it\n        if instance_name in self.jobs_results:\n            self.log.debug(\"Instance: %s cleaned from jobs results.\" % instance_name)\n            ret = self.jobs_results[instance_name].get()\n            if isinstance(ret, Exception):\n                self.log.exception(\"Exception in worker thread: {0}\".format(ret))\n            del self.jobs_results[instance_name]\n\n\n    def _check(self, instance):\n        \"\"\"This function should be implemented by inherited classes\"\"\"\n        raise NotImplementedError\n\n\n    def _clean(self):\n        now = time.time()\n        for name, start_time in self.jobs_status.iteritems():\n            if now - start_time > TIMEOUT:\n                self.log.critical(\"Restarting Pool. One check is stuck: %s\" % name)\n                self.restart_pool()\n                break\n", "path": "checks/network_checks.py"}]}