We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
 Importing ABC directly from collections will be removed in Python 3.10
Use collections for Python 2 and collections.abc for Python 3

https://github.com/DataDog/integrations-extras/blob/7e676a2d256e142ddf4cb44ac11a6e590a502779/filebeat/datadog_checks/filebeat/filebeat.py#L148

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `filebeat/datadog_checks/filebeat/filebeat.py`
Content:
```
1 # (C) Datadog, Inc. 2010-2016
2 # All rights reserved
3 # Licensed under Simplified BSD License (see LICENSE)
4 
5 # stdlib
6 import collections
7 import errno
8 import json
9 import numbers
10 import os
11 import re
12 import sre_constants
13 
14 import requests
15 from six import iteritems
16 
17 from datadog_checks.base import AgentCheck, is_affirmative
18 from datadog_checks.base.utils.containers import hash_mutable
19 
20 EVENT_TYPE = SOURCE_TYPE_NAME = "filebeat"
21 
22 
23 class FilebeatCheckHttpProfiler:
24     """
25     Filebeat's HTTP profiler gives a bunch of counter variables; their value holds little interest,
26     what we really want is the delta in between runs. This class is responsible for caching the
27     values from the previous run
28     """
29 
30     INCREMENT_METRIC_NAMES = [
31         "filebeat.events.done",
32         "filebeat.harvester.closed",
33         "filebeat.harvester.files.truncated",
34         "filebeat.harvester.open_files",
35         "filebeat.harvester.skipped",
36         "filebeat.harvester.started",
37         "filebeat.prospector.log.files.renamed",
38         "filebeat.prospector.log.files.truncated",
39         "libbeat.config.module.running",
40         "libbeat.config.module.starts",
41         "libbeat.config.module.stops",
42         "libbeat.config.reloads",
43         "libbeat.es.call_count.PublishEvents",
44         "libbeat.es.publish.read_bytes",
45         "libbeat.es.publish.read_errors",
46         "libbeat.es.publish.write_bytes",
47         "libbeat.es.publish.write_errors",
48         "libbeat.es.published_and_acked_events",
49         "libbeat.es.published_but_not_acked_events",
50         "libbeat.kafka.call_count.PublishEvents",
51         "libbeat.kafka.published_and_acked_events",
52         "libbeat.kafka.published_but_not_acked_events",
53         "libbeat.logstash.call_count.PublishEvents",
54         "libbeat.logstash.publish.read_bytes",
55         "libbeat.logstash.publish.read_errors",
56         "libbeat.logstash.publish.write_bytes",
57         "libbeat.logstash.publish.write_errors",
58         "libbeat.logstash.published_and_acked_events",
59         "libbeat.logstash.published_but_not_acked_events",
60         "libbeat.output.events.acked",
61         "libbeat.output.events.dropped",
62         "libbeat.output.events.failed",
63         "libbeat.output.events.total",
64         "libbeat.pipeline.events.dropped",
65         "libbeat.pipeline.events.failed",
66         "libbeat.pipeline.events.filtered",
67         "libbeat.pipeline.events.published",
68         "libbeat.pipeline.events.total",
69         "libbeat.publisher.messages_in_worker_queues",
70         "libbeat.publisher.published_events",
71         "libbeat.redis.publish.read_bytes",
72         "libbeat.redis.publish.read_errors",
73         "libbeat.redis.publish.write_bytes",
74         "libbeat.redis.publish.write_errors",
75         "publish.events",
76         "registrar.states.cleanup",
77         "registrar.states.current",
78         "registrar.states.update",
79         "registrar.writes",
80     ]
81 
82     GAUGE_METRIC_NAMES = ["filebeat.harvester.running"]
83 
84     VARS_ROUTE = "debug/vars"
85 
86     def __init__(self, config):
87         self._config = config
88         self._previous_increment_values = {}
89         # regex matching ain't free, let's cache this
90         self._should_keep_metrics = {}
91 
92     def gather_metrics(self):
93         if not self._config.stats_endpoint:
94             return {}
95 
96         response = self._make_request()
97 
98         return {"increment": self._gather_increment_metrics(response), "gauge": self._gather_gauge_metrics(response)}
99 
100     def _make_request(self):
101 
102         response = requests.get(self._config.stats_endpoint, timeout=self._config.timeout)
103         response.raise_for_status()
104 
105         return self.flatten(response.json())
106 
107     def _gather_increment_metrics(self, response):
108         new_values = {
109             name: response[name]
110             for name in self.INCREMENT_METRIC_NAMES
111             if self._should_keep_metric(name) and name in response
112         }
113 
114         deltas = self._compute_increment_deltas(new_values)
115 
116         self._previous_increment_values = new_values
117 
118         return deltas
119 
120     def _compute_increment_deltas(self, new_values):
121         deltas = {}
122 
123         for name, new_value in iteritems(new_values):
124             if name not in self._previous_increment_values or self._previous_increment_values[name] > new_value:
125                 # either the agent or filebeat got restarted, we're not
126                 # reporting anything this time around
127                 return {}
128             deltas[name] = new_value - self._previous_increment_values[name]
129 
130         return deltas
131 
132     def _gather_gauge_metrics(self, response):
133         return {
134             name: response[name]
135             for name in self.GAUGE_METRIC_NAMES
136             if self._should_keep_metric(name) and name in response
137         }
138 
139     def _should_keep_metric(self, name):
140         if name not in self._should_keep_metrics:
141             self._should_keep_metrics[name] = self._config.should_keep_metric(name)
142         return self._should_keep_metrics[name]
143 
144     def flatten(self, d, parent_key="", sep="."):
145         items = []
146         for k, v in d.items():
147             new_key = parent_key + sep + k if parent_key else k
148             if isinstance(v, collections.MutableMapping):
149                 items.extend(self.flatten(v, new_key, sep=sep).items())
150             else:
151                 items.append((new_key, v))
152         return dict(items)
153 
154 
155 class FilebeatCheckInstanceConfig:
156 
157     _only_metrics_regexes = None
158 
159     def __init__(self, instance):
160         self._registry_file_path = instance.get("registry_file_path")
161         if self._registry_file_path is None:
162             raise Exception("An absolute path to a filebeat registry path must be specified")
163 
164         self._stats_endpoint = instance.get("stats_endpoint")
165 
166         self._only_metrics = instance.get("only_metrics", [])
167 
168         if not isinstance(self._only_metrics, list):
169             raise Exception(
170                 "If given, filebeat's only_metrics must be a list of regexes, got %s" % (self._only_metrics,)
171             )
172 
173         self._timeout = instance.get("timeout", 2)
174         if not isinstance(self._timeout, numbers.Real) or self._timeout <= 0:
175             raise Exception("If given, filebeats timeout must be a positive number, got %s" % (self._timeout,))
176 
177     @property
178     def registry_file_path(self):
179         return self._registry_file_path
180 
181     @property
182     def stats_endpoint(self):
183         return self._stats_endpoint
184 
185     @property
186     def timeout(self):
187         return self._timeout
188 
189     def should_keep_metric(self, metric_name):
190 
191         if not self._only_metrics:
192             return True
193 
194         return any(re.search(regex, metric_name) for regex in self._compiled_regexes())
195 
196     def _compiled_regexes(self):
197         if self._only_metrics_regexes is None:
198             self._only_metrics_regexes = self._compile_regexes()
199         return self._only_metrics_regexes
200 
201     def _compile_regexes(self):
202         compiled_regexes = []
203 
204         for regex in self._only_metrics:
205             try:
206                 compiled_regexes.append(re.compile(regex))
207             except sre_constants.error as ex:
208                 raise Exception('Invalid only_metric regex for filebeat: "%s", error: %s' % (regex, ex))
209 
210         return compiled_regexes
211 
212 
213 class FilebeatCheck(AgentCheck):
214 
215     METRIC_PREFIX = "filebeat."
216 
217     def __init__(self, *args, **kwargs):
218         AgentCheck.__init__(self, *args, **kwargs)
219         self.instance_cache = {}
220 
221     def check(self, instance):
222         normalize_metrics = is_affirmative(instance.get("normalize_metrics", False))
223 
224         instance_key = hash_mutable(instance)
225         if instance_key in self.instance_cache:
226             config = self.instance_cache[instance_key]["config"]
227             profiler = self.instance_cache[instance_key]["profiler"]
228         else:
229             config = FilebeatCheckInstanceConfig(instance)
230             profiler = FilebeatCheckHttpProfiler(config)
231             self.instance_cache[instance_key] = {"config": config, "profiler": profiler}
232 
233         self._process_registry(config)
234         self._gather_http_profiler_metrics(config, profiler, normalize_metrics)
235 
236     def _process_registry(self, config):
237         registry_contents = self._parse_registry_file(config.registry_file_path)
238 
239         if isinstance(registry_contents, dict):
240             # filebeat version < 5
241             registry_contents = registry_contents.values()
242 
243         for item in registry_contents:
244             self._process_registry_item(item)
245 
246     def _parse_registry_file(self, registry_file_path):
247         try:
248             with open(registry_file_path) as registry_file:
249                 return json.load(registry_file)
250         except IOError as ex:
251             self.log.error("Cannot read the registry log file at %s: %s", registry_file_path, ex)
252 
253             if ex.errno == errno.EACCES:
254                 self.log.error(
255                     "You might be interesting in having a look at " "https://github.com/elastic/beats/pull/6455"
256                 )
257 
258             return []
259 
260     def _process_registry_item(self, item):
261         source = item["source"]
262         offset = item["offset"]
263 
264         try:
265             stats = os.stat(source)
266 
267             if self._is_same_file(stats, item["FileStateOS"]):
268                 unprocessed_bytes = stats.st_size - offset
269 
270                 self.gauge("filebeat.registry.unprocessed_bytes", unprocessed_bytes, tags=["source:{0}".format(source)])
271             else:
272                 self.log.debug("Filebeat source %s appears to have changed", source)
273         except OSError:
274             self.log.debug("Unable to get stats on filebeat source %s", source)
275 
276     def _is_same_file(self, stats, file_state_os):
277         return stats.st_dev == file_state_os["device"] and stats.st_ino == file_state_os["inode"]
278 
279     def _gather_http_profiler_metrics(self, config, profiler, normalize_metrics):
280         try:
281             all_metrics = profiler.gather_metrics()
282         except Exception as ex:
283             self.log.error("Error when fetching metrics from %s: %s", config.stats_endpoint, ex)
284             return
285 
286         tags = ["stats_endpoint:{0}".format(config.stats_endpoint)]
287 
288         for action, metrics in iteritems(all_metrics):
289             method = getattr(self, action)
290 
291             for name, value in iteritems(metrics):
292                 if not name.startswith(self.METRIC_PREFIX) and normalize_metrics:
293                     name = self.METRIC_PREFIX + name
294                 method(name, value, tags)
295
```
--- 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/filebeat/datadog_checks/filebeat/filebeat.py b/filebeat/datadog_checks/filebeat/filebeat.py
--- a/filebeat/datadog_checks/filebeat/filebeat.py
+++ b/filebeat/datadog_checks/filebeat/filebeat.py
@@ -3,7 +3,6 @@
 # Licensed under Simplified BSD License (see LICENSE)
 
 # stdlib
-import collections
 import errno
 import json
 import numbers
@@ -12,11 +11,17 @@
 import sre_constants
 
 import requests
+import six
 from six import iteritems
 
 from datadog_checks.base import AgentCheck, is_affirmative
 from datadog_checks.base.utils.containers import hash_mutable
 
+if six.PY3:
+    from collections.abc import MutableMapping
+else:
+    from collections import MutableMapping
+
 EVENT_TYPE = SOURCE_TYPE_NAME = "filebeat"
 
 
@@ -145,7 +150,7 @@
         items = []
         for k, v in d.items():
             new_key = parent_key + sep + k if parent_key else k
-            if isinstance(v, collections.MutableMapping):
+            if isinstance(v, MutableMapping):
                 items.extend(self.flatten(v, new_key, sep=sep).items())
             else:
                 items.append((new_key, v))

{"golden_diff": "diff --git a/filebeat/datadog_checks/filebeat/filebeat.py b/filebeat/datadog_checks/filebeat/filebeat.py\n--- a/filebeat/datadog_checks/filebeat/filebeat.py\n+++ b/filebeat/datadog_checks/filebeat/filebeat.py\n@@ -3,7 +3,6 @@\n # Licensed under Simplified BSD License (see LICENSE)\n \n # stdlib\n-import collections\n import errno\n import json\n import numbers\n@@ -12,11 +11,17 @@\n import sre_constants\n \n import requests\n+import six\n from six import iteritems\n \n from datadog_checks.base import AgentCheck, is_affirmative\n from datadog_checks.base.utils.containers import hash_mutable\n \n+if six.PY3:\n+    from collections.abc import MutableMapping\n+else:\n+    from collections import MutableMapping\n+\n EVENT_TYPE = SOURCE_TYPE_NAME = \"filebeat\"\n \n \n@@ -145,7 +150,7 @@\n         items = []\n         for k, v in d.items():\n             new_key = parent_key + sep + k if parent_key else k\n-            if isinstance(v, collections.MutableMapping):\n+            if isinstance(v, MutableMapping):\n                 items.extend(self.flatten(v, new_key, sep=sep).items())\n             else:\n                 items.append((new_key, v))\n", "issue": " Importing ABC directly from collections will be removed in Python 3.10\nUse collections for Python 2 and collections.abc for Python 3\r\n\r\nhttps://github.com/DataDog/integrations-extras/blob/7e676a2d256e142ddf4cb44ac11a6e590a502779/filebeat/datadog_checks/filebeat/filebeat.py#L148\n", "before_files": [{"content": "# (C) Datadog, Inc. 2010-2016\n# All rights reserved\n# Licensed under Simplified BSD License (see LICENSE)\n\n# stdlib\nimport collections\nimport errno\nimport json\nimport numbers\nimport os\nimport re\nimport sre_constants\n\nimport requests\nfrom six import iteritems\n\nfrom datadog_checks.base import AgentCheck, is_affirmative\nfrom datadog_checks.base.utils.containers import hash_mutable\n\nEVENT_TYPE = SOURCE_TYPE_NAME = \"filebeat\"\n\n\nclass FilebeatCheckHttpProfiler:\n    \"\"\"\n    Filebeat's HTTP profiler gives a bunch of counter variables; their value holds little interest,\n    what we really want is the delta in between runs. This class is responsible for caching the\n    values from the previous run\n    \"\"\"\n\n    INCREMENT_METRIC_NAMES = [\n        \"filebeat.events.done\",\n        \"filebeat.harvester.closed\",\n        \"filebeat.harvester.files.truncated\",\n        \"filebeat.harvester.open_files\",\n        \"filebeat.harvester.skipped\",\n        \"filebeat.harvester.started\",\n        \"filebeat.prospector.log.files.renamed\",\n        \"filebeat.prospector.log.files.truncated\",\n        \"libbeat.config.module.running\",\n        \"libbeat.config.module.starts\",\n        \"libbeat.config.module.stops\",\n        \"libbeat.config.reloads\",\n        \"libbeat.es.call_count.PublishEvents\",\n        \"libbeat.es.publish.read_bytes\",\n        \"libbeat.es.publish.read_errors\",\n        \"libbeat.es.publish.write_bytes\",\n        \"libbeat.es.publish.write_errors\",\n        \"libbeat.es.published_and_acked_events\",\n        \"libbeat.es.published_but_not_acked_events\",\n        \"libbeat.kafka.call_count.PublishEvents\",\n        \"libbeat.kafka.published_and_acked_events\",\n        \"libbeat.kafka.published_but_not_acked_events\",\n        \"libbeat.logstash.call_count.PublishEvents\",\n        \"libbeat.logstash.publish.read_bytes\",\n        \"libbeat.logstash.publish.read_errors\",\n        \"libbeat.logstash.publish.write_bytes\",\n        \"libbeat.logstash.publish.write_errors\",\n        \"libbeat.logstash.published_and_acked_events\",\n        \"libbeat.logstash.published_but_not_acked_events\",\n        \"libbeat.output.events.acked\",\n        \"libbeat.output.events.dropped\",\n        \"libbeat.output.events.failed\",\n        \"libbeat.output.events.total\",\n        \"libbeat.pipeline.events.dropped\",\n        \"libbeat.pipeline.events.failed\",\n        \"libbeat.pipeline.events.filtered\",\n        \"libbeat.pipeline.events.published\",\n        \"libbeat.pipeline.events.total\",\n        \"libbeat.publisher.messages_in_worker_queues\",\n        \"libbeat.publisher.published_events\",\n        \"libbeat.redis.publish.read_bytes\",\n        \"libbeat.redis.publish.read_errors\",\n        \"libbeat.redis.publish.write_bytes\",\n        \"libbeat.redis.publish.write_errors\",\n        \"publish.events\",\n        \"registrar.states.cleanup\",\n        \"registrar.states.current\",\n        \"registrar.states.update\",\n        \"registrar.writes\",\n    ]\n\n    GAUGE_METRIC_NAMES = [\"filebeat.harvester.running\"]\n\n    VARS_ROUTE = \"debug/vars\"\n\n    def __init__(self, config):\n        self._config = config\n        self._previous_increment_values = {}\n        # regex matching ain't free, let's cache this\n        self._should_keep_metrics = {}\n\n    def gather_metrics(self):\n        if not self._config.stats_endpoint:\n            return {}\n\n        response = self._make_request()\n\n        return {\"increment\": self._gather_increment_metrics(response), \"gauge\": self._gather_gauge_metrics(response)}\n\n    def _make_request(self):\n\n        response = requests.get(self._config.stats_endpoint, timeout=self._config.timeout)\n        response.raise_for_status()\n\n        return self.flatten(response.json())\n\n    def _gather_increment_metrics(self, response):\n        new_values = {\n            name: response[name]\n            for name in self.INCREMENT_METRIC_NAMES\n            if self._should_keep_metric(name) and name in response\n        }\n\n        deltas = self._compute_increment_deltas(new_values)\n\n        self._previous_increment_values = new_values\n\n        return deltas\n\n    def _compute_increment_deltas(self, new_values):\n        deltas = {}\n\n        for name, new_value in iteritems(new_values):\n            if name not in self._previous_increment_values or self._previous_increment_values[name] > new_value:\n                # either the agent or filebeat got restarted, we're not\n                # reporting anything this time around\n                return {}\n            deltas[name] = new_value - self._previous_increment_values[name]\n\n        return deltas\n\n    def _gather_gauge_metrics(self, response):\n        return {\n            name: response[name]\n            for name in self.GAUGE_METRIC_NAMES\n            if self._should_keep_metric(name) and name in response\n        }\n\n    def _should_keep_metric(self, name):\n        if name not in self._should_keep_metrics:\n            self._should_keep_metrics[name] = self._config.should_keep_metric(name)\n        return self._should_keep_metrics[name]\n\n    def flatten(self, d, parent_key=\"\", sep=\".\"):\n        items = []\n        for k, v in d.items():\n            new_key = parent_key + sep + k if parent_key else k\n            if isinstance(v, collections.MutableMapping):\n                items.extend(self.flatten(v, new_key, sep=sep).items())\n            else:\n                items.append((new_key, v))\n        return dict(items)\n\n\nclass FilebeatCheckInstanceConfig:\n\n    _only_metrics_regexes = None\n\n    def __init__(self, instance):\n        self._registry_file_path = instance.get(\"registry_file_path\")\n        if self._registry_file_path is None:\n            raise Exception(\"An absolute path to a filebeat registry path must be specified\")\n\n        self._stats_endpoint = instance.get(\"stats_endpoint\")\n\n        self._only_metrics = instance.get(\"only_metrics\", [])\n\n        if not isinstance(self._only_metrics, list):\n            raise Exception(\n                \"If given, filebeat's only_metrics must be a list of regexes, got %s\" % (self._only_metrics,)\n            )\n\n        self._timeout = instance.get(\"timeout\", 2)\n        if not isinstance(self._timeout, numbers.Real) or self._timeout <= 0:\n            raise Exception(\"If given, filebeats timeout must be a positive number, got %s\" % (self._timeout,))\n\n    @property\n    def registry_file_path(self):\n        return self._registry_file_path\n\n    @property\n    def stats_endpoint(self):\n        return self._stats_endpoint\n\n    @property\n    def timeout(self):\n        return self._timeout\n\n    def should_keep_metric(self, metric_name):\n\n        if not self._only_metrics:\n            return True\n\n        return any(re.search(regex, metric_name) for regex in self._compiled_regexes())\n\n    def _compiled_regexes(self):\n        if self._only_metrics_regexes is None:\n            self._only_metrics_regexes = self._compile_regexes()\n        return self._only_metrics_regexes\n\n    def _compile_regexes(self):\n        compiled_regexes = []\n\n        for regex in self._only_metrics:\n            try:\n                compiled_regexes.append(re.compile(regex))\n            except sre_constants.error as ex:\n                raise Exception('Invalid only_metric regex for filebeat: \"%s\", error: %s' % (regex, ex))\n\n        return compiled_regexes\n\n\nclass FilebeatCheck(AgentCheck):\n\n    METRIC_PREFIX = \"filebeat.\"\n\n    def __init__(self, *args, **kwargs):\n        AgentCheck.__init__(self, *args, **kwargs)\n        self.instance_cache = {}\n\n    def check(self, instance):\n        normalize_metrics = is_affirmative(instance.get(\"normalize_metrics\", False))\n\n        instance_key = hash_mutable(instance)\n        if instance_key in self.instance_cache:\n            config = self.instance_cache[instance_key][\"config\"]\n            profiler = self.instance_cache[instance_key][\"profiler\"]\n        else:\n            config = FilebeatCheckInstanceConfig(instance)\n            profiler = FilebeatCheckHttpProfiler(config)\n            self.instance_cache[instance_key] = {\"config\": config, \"profiler\": profiler}\n\n        self._process_registry(config)\n        self._gather_http_profiler_metrics(config, profiler, normalize_metrics)\n\n    def _process_registry(self, config):\n        registry_contents = self._parse_registry_file(config.registry_file_path)\n\n        if isinstance(registry_contents, dict):\n            # filebeat version < 5\n            registry_contents = registry_contents.values()\n\n        for item in registry_contents:\n            self._process_registry_item(item)\n\n    def _parse_registry_file(self, registry_file_path):\n        try:\n            with open(registry_file_path) as registry_file:\n                return json.load(registry_file)\n        except IOError as ex:\n            self.log.error(\"Cannot read the registry log file at %s: %s\", registry_file_path, ex)\n\n            if ex.errno == errno.EACCES:\n                self.log.error(\n                    \"You might be interesting in having a look at \" \"https://github.com/elastic/beats/pull/6455\"\n                )\n\n            return []\n\n    def _process_registry_item(self, item):\n        source = item[\"source\"]\n        offset = item[\"offset\"]\n\n        try:\n            stats = os.stat(source)\n\n            if self._is_same_file(stats, item[\"FileStateOS\"]):\n                unprocessed_bytes = stats.st_size - offset\n\n                self.gauge(\"filebeat.registry.unprocessed_bytes\", unprocessed_bytes, tags=[\"source:{0}\".format(source)])\n            else:\n                self.log.debug(\"Filebeat source %s appears to have changed\", source)\n        except OSError:\n            self.log.debug(\"Unable to get stats on filebeat source %s\", source)\n\n    def _is_same_file(self, stats, file_state_os):\n        return stats.st_dev == file_state_os[\"device\"] and stats.st_ino == file_state_os[\"inode\"]\n\n    def _gather_http_profiler_metrics(self, config, profiler, normalize_metrics):\n        try:\n            all_metrics = profiler.gather_metrics()\n        except Exception as ex:\n            self.log.error(\"Error when fetching metrics from %s: %s\", config.stats_endpoint, ex)\n            return\n\n        tags = [\"stats_endpoint:{0}\".format(config.stats_endpoint)]\n\n        for action, metrics in iteritems(all_metrics):\n            method = getattr(self, action)\n\n            for name, value in iteritems(metrics):\n                if not name.startswith(self.METRIC_PREFIX) and normalize_metrics:\n                    name = self.METRIC_PREFIX + name\n                method(name, value, tags)\n", "path": "filebeat/datadog_checks/filebeat/filebeat.py"}], "after_files": [{"content": "# (C) Datadog, Inc. 2010-2016\n# All rights reserved\n# Licensed under Simplified BSD License (see LICENSE)\n\n# stdlib\nimport errno\nimport json\nimport numbers\nimport os\nimport re\nimport sre_constants\n\nimport requests\nimport six\nfrom six import iteritems\n\nfrom datadog_checks.base import AgentCheck, is_affirmative\nfrom datadog_checks.base.utils.containers import hash_mutable\n\nif six.PY3:\n    from collections.abc import MutableMapping\nelse:\n    from collections import MutableMapping\n\nEVENT_TYPE = SOURCE_TYPE_NAME = \"filebeat\"\n\n\nclass FilebeatCheckHttpProfiler:\n    \"\"\"\n    Filebeat's HTTP profiler gives a bunch of counter variables; their value holds little interest,\n    what we really want is the delta in between runs. This class is responsible for caching the\n    values from the previous run\n    \"\"\"\n\n    INCREMENT_METRIC_NAMES = [\n        \"filebeat.events.done\",\n        \"filebeat.harvester.closed\",\n        \"filebeat.harvester.files.truncated\",\n        \"filebeat.harvester.open_files\",\n        \"filebeat.harvester.skipped\",\n        \"filebeat.harvester.started\",\n        \"filebeat.prospector.log.files.renamed\",\n        \"filebeat.prospector.log.files.truncated\",\n        \"libbeat.config.module.running\",\n        \"libbeat.config.module.starts\",\n        \"libbeat.config.module.stops\",\n        \"libbeat.config.reloads\",\n        \"libbeat.es.call_count.PublishEvents\",\n        \"libbeat.es.publish.read_bytes\",\n        \"libbeat.es.publish.read_errors\",\n        \"libbeat.es.publish.write_bytes\",\n        \"libbeat.es.publish.write_errors\",\n        \"libbeat.es.published_and_acked_events\",\n        \"libbeat.es.published_but_not_acked_events\",\n        \"libbeat.kafka.call_count.PublishEvents\",\n        \"libbeat.kafka.published_and_acked_events\",\n        \"libbeat.kafka.published_but_not_acked_events\",\n        \"libbeat.logstash.call_count.PublishEvents\",\n        \"libbeat.logstash.publish.read_bytes\",\n        \"libbeat.logstash.publish.read_errors\",\n        \"libbeat.logstash.publish.write_bytes\",\n        \"libbeat.logstash.publish.write_errors\",\n        \"libbeat.logstash.published_and_acked_events\",\n        \"libbeat.logstash.published_but_not_acked_events\",\n        \"libbeat.output.events.acked\",\n        \"libbeat.output.events.dropped\",\n        \"libbeat.output.events.failed\",\n        \"libbeat.output.events.total\",\n        \"libbeat.pipeline.events.dropped\",\n        \"libbeat.pipeline.events.failed\",\n        \"libbeat.pipeline.events.filtered\",\n        \"libbeat.pipeline.events.published\",\n        \"libbeat.pipeline.events.total\",\n        \"libbeat.publisher.messages_in_worker_queues\",\n        \"libbeat.publisher.published_events\",\n        \"libbeat.redis.publish.read_bytes\",\n        \"libbeat.redis.publish.read_errors\",\n        \"libbeat.redis.publish.write_bytes\",\n        \"libbeat.redis.publish.write_errors\",\n        \"publish.events\",\n        \"registrar.states.cleanup\",\n        \"registrar.states.current\",\n        \"registrar.states.update\",\n        \"registrar.writes\",\n    ]\n\n    GAUGE_METRIC_NAMES = [\"filebeat.harvester.running\"]\n\n    VARS_ROUTE = \"debug/vars\"\n\n    def __init__(self, config):\n        self._config = config\n        self._previous_increment_values = {}\n        # regex matching ain't free, let's cache this\n        self._should_keep_metrics = {}\n\n    def gather_metrics(self):\n        if not self._config.stats_endpoint:\n            return {}\n\n        response = self._make_request()\n\n        return {\"increment\": self._gather_increment_metrics(response), \"gauge\": self._gather_gauge_metrics(response)}\n\n    def _make_request(self):\n\n        response = requests.get(self._config.stats_endpoint, timeout=self._config.timeout)\n        response.raise_for_status()\n\n        return self.flatten(response.json())\n\n    def _gather_increment_metrics(self, response):\n        new_values = {\n            name: response[name]\n            for name in self.INCREMENT_METRIC_NAMES\n            if self._should_keep_metric(name) and name in response\n        }\n\n        deltas = self._compute_increment_deltas(new_values)\n\n        self._previous_increment_values = new_values\n\n        return deltas\n\n    def _compute_increment_deltas(self, new_values):\n        deltas = {}\n\n        for name, new_value in iteritems(new_values):\n            if name not in self._previous_increment_values or self._previous_increment_values[name] > new_value:\n                # either the agent or filebeat got restarted, we're not\n                # reporting anything this time around\n                return {}\n            deltas[name] = new_value - self._previous_increment_values[name]\n\n        return deltas\n\n    def _gather_gauge_metrics(self, response):\n        return {\n            name: response[name]\n            for name in self.GAUGE_METRIC_NAMES\n            if self._should_keep_metric(name) and name in response\n        }\n\n    def _should_keep_metric(self, name):\n        if name not in self._should_keep_metrics:\n            self._should_keep_metrics[name] = self._config.should_keep_metric(name)\n        return self._should_keep_metrics[name]\n\n    def flatten(self, d, parent_key=\"\", sep=\".\"):\n        items = []\n        for k, v in d.items():\n            new_key = parent_key + sep + k if parent_key else k\n            if isinstance(v, MutableMapping):\n                items.extend(self.flatten(v, new_key, sep=sep).items())\n            else:\n                items.append((new_key, v))\n        return dict(items)\n\n\nclass FilebeatCheckInstanceConfig:\n\n    _only_metrics_regexes = None\n\n    def __init__(self, instance):\n        self._registry_file_path = instance.get(\"registry_file_path\")\n        if self._registry_file_path is None:\n            raise Exception(\"An absolute path to a filebeat registry path must be specified\")\n\n        self._stats_endpoint = instance.get(\"stats_endpoint\")\n\n        self._only_metrics = instance.get(\"only_metrics\", [])\n\n        if not isinstance(self._only_metrics, list):\n            raise Exception(\n                \"If given, filebeat's only_metrics must be a list of regexes, got %s\" % (self._only_metrics,)\n            )\n\n        self._timeout = instance.get(\"timeout\", 2)\n        if not isinstance(self._timeout, numbers.Real) or self._timeout <= 0:\n            raise Exception(\"If given, filebeats timeout must be a positive number, got %s\" % (self._timeout,))\n\n    @property\n    def registry_file_path(self):\n        return self._registry_file_path\n\n    @property\n    def stats_endpoint(self):\n        return self._stats_endpoint\n\n    @property\n    def timeout(self):\n        return self._timeout\n\n    def should_keep_metric(self, metric_name):\n\n        if not self._only_metrics:\n            return True\n\n        return any(re.search(regex, metric_name) for regex in self._compiled_regexes())\n\n    def _compiled_regexes(self):\n        if self._only_metrics_regexes is None:\n            self._only_metrics_regexes = self._compile_regexes()\n        return self._only_metrics_regexes\n\n    def _compile_regexes(self):\n        compiled_regexes = []\n\n        for regex in self._only_metrics:\n            try:\n                compiled_regexes.append(re.compile(regex))\n            except sre_constants.error as ex:\n                raise Exception('Invalid only_metric regex for filebeat: \"%s\", error: %s' % (regex, ex))\n\n        return compiled_regexes\n\n\nclass FilebeatCheck(AgentCheck):\n\n    METRIC_PREFIX = \"filebeat.\"\n\n    def __init__(self, *args, **kwargs):\n        AgentCheck.__init__(self, *args, **kwargs)\n        self.instance_cache = {}\n\n    def check(self, instance):\n        normalize_metrics = is_affirmative(instance.get(\"normalize_metrics\", False))\n\n        instance_key = hash_mutable(instance)\n        if instance_key in self.instance_cache:\n            config = self.instance_cache[instance_key][\"config\"]\n            profiler = self.instance_cache[instance_key][\"profiler\"]\n        else:\n            config = FilebeatCheckInstanceConfig(instance)\n            profiler = FilebeatCheckHttpProfiler(config)\n            self.instance_cache[instance_key] = {\"config\": config, \"profiler\": profiler}\n\n        self._process_registry(config)\n        self._gather_http_profiler_metrics(config, profiler, normalize_metrics)\n\n    def _process_registry(self, config):\n        registry_contents = self._parse_registry_file(config.registry_file_path)\n\n        if isinstance(registry_contents, dict):\n            # filebeat version < 5\n            registry_contents = registry_contents.values()\n\n        for item in registry_contents:\n            self._process_registry_item(item)\n\n    def _parse_registry_file(self, registry_file_path):\n        try:\n            with open(registry_file_path) as registry_file:\n                return json.load(registry_file)\n        except IOError as ex:\n            self.log.error(\"Cannot read the registry log file at %s: %s\", registry_file_path, ex)\n\n            if ex.errno == errno.EACCES:\n                self.log.error(\n                    \"You might be interesting in having a look at \" \"https://github.com/elastic/beats/pull/6455\"\n                )\n\n            return []\n\n    def _process_registry_item(self, item):\n        source = item[\"source\"]\n        offset = item[\"offset\"]\n\n        try:\n            stats = os.stat(source)\n\n            if self._is_same_file(stats, item[\"FileStateOS\"]):\n                unprocessed_bytes = stats.st_size - offset\n\n                self.gauge(\"filebeat.registry.unprocessed_bytes\", unprocessed_bytes, tags=[\"source:{0}\".format(source)])\n            else:\n                self.log.debug(\"Filebeat source %s appears to have changed\", source)\n        except OSError:\n            self.log.debug(\"Unable to get stats on filebeat source %s\", source)\n\n    def _is_same_file(self, stats, file_state_os):\n        return stats.st_dev == file_state_os[\"device\"] and stats.st_ino == file_state_os[\"inode\"]\n\n    def _gather_http_profiler_metrics(self, config, profiler, normalize_metrics):\n        try:\n            all_metrics = profiler.gather_metrics()\n        except Exception as ex:\n            self.log.error(\"Error when fetching metrics from %s: %s\", config.stats_endpoint, ex)\n            return\n\n        tags = [\"stats_endpoint:{0}\".format(config.stats_endpoint)]\n\n        for action, metrics in iteritems(all_metrics):\n            method = getattr(self, action)\n\n            for name, value in iteritems(metrics):\n                if not name.startswith(self.METRIC_PREFIX) and normalize_metrics:\n                    name = self.METRIC_PREFIX + name\n                method(name, value, tags)\n", "path": "filebeat/datadog_checks/filebeat/filebeat.py"}]}