We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add support for filebeat version 7.x
**Docker image:** custom docker image based on `datadog/agent:7.28.0` with  this added to the `Dockerfile` `RUN agent integration install -r -t datadog-filebeat==1.0.0`

Since filebeat version `7.x`, the registry file is moved to [directory format][0]. Therefore, the current config will not work as the registry is not in a single file format anymore.

[Snippet][1] from the config example for filebeat:
```
instances:

    ## @param registry_file_path - string - required
    ## The absolute path to the registry file used by Filebeat.
    ##
    ## See https://www.elastic.co/guide/en/beats/filebeat/current/migration-registry-file.html
    #
  - registry_file_path: /var/lib/filebeat/registry
```

The error message from `docker logs datadog-agent`
```
2021-06-17 03:20:33 UTC | CORE | ERROR | (pkg/collector/python/datadog_agent.go:120 in LogMessage) | filebeat:71abd2e9bec66d7b | (filebeat.py:251) | Cannot read the registry log file at /usr/share/filebeat/data/registry: [Errno 21] Is a directory: '/usr/share/filebeat/data/registry'
```

So datadog-agent  and filebeat integration will not work with filebeat version 7.x for now.


[1]: https://github.com/DataDog/integrations-extras/blob/master/filebeat/datadog_checks/filebeat/data/conf.yaml.example#L46-L53
[0]: https://www.elastic.co/guide/en/beats/filebeat/current/configuration-general-options.html#_registry_migrate_file


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

{"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@@ -261,6 +261,16 @@\n         try:\n             with open(registry_file_path) as registry_file:\n                 return json.load(registry_file)\n+\n+        except ValueError:\n+            contents = []\n+            with open(registry_file_path) as registry_file:\n+                for line in registry_file:\n+                    content = json.loads(line)\n+                    if 'op' not in content.keys():\n+                        contents.append(content['v'])\n+            return contents\n+\n         except IOError as ex:\n             self.log.error(\"Cannot read the registry log file at %s: %s\", registry_file_path, ex)\n", "issue": "Add support for filebeat version 7.x\n**Docker image:** custom docker image based on `datadog/agent:7.28.0` with  this added to the `Dockerfile` `RUN agent integration install -r -t datadog-filebeat==1.0.0`\r\n\r\nSince filebeat version `7.x`, the registry file is moved to [directory format][0]. Therefore, the current config will not work as the registry is not in a single file format anymore.\r\n\r\n[Snippet][1] from the config example for filebeat:\r\n```\r\ninstances:\r\n\r\n    ## @param registry_file_path - string - required\r\n    ## The absolute path to the registry file used by Filebeat.\r\n    ##\r\n    ## See https://www.elastic.co/guide/en/beats/filebeat/current/migration-registry-file.html\r\n    #\r\n  - registry_file_path: /var/lib/filebeat/registry\r\n```\r\n\r\nThe error message from `docker logs datadog-agent`\r\n```\r\n2021-06-17 03:20:33 UTC | CORE | ERROR | (pkg/collector/python/datadog_agent.go:120 in LogMessage) | filebeat:71abd2e9bec66d7b | (filebeat.py:251) | Cannot read the registry log file at /usr/share/filebeat/data/registry: [Errno 21] Is a directory: '/usr/share/filebeat/data/registry'\r\n```\r\n\r\nSo datadog-agent  and filebeat integration will not work with filebeat version 7.x for now.\r\n\r\n\r\n[1]: https://github.com/DataDog/integrations-extras/blob/master/filebeat/datadog_checks/filebeat/data/conf.yaml.example#L46-L53\r\n[0]: https://www.elastic.co/guide/en/beats/filebeat/current/configuration-general-options.html#_registry_migrate_file\r\n\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 errno\nimport json\nimport os\nimport re\nimport sre_constants\n\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, http):\n        self._config = config\n        self._http = http\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 = self._http.get(self._config.stats_endpoint)\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        self._ignore_registry = instance.get(\"ignore_registry\", False)\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    @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 ignore_registry(self):\n        return self._ignore_registry\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    SERVICE_CHECK_NAME = 'can_connect'\n\n    __NAMESPACE__ = \"filebeat\"\n\n    def __init__(self, *args, **kwargs):\n        AgentCheck.__init__(self, *args, **kwargs)\n        self.instance_cache = {}\n        self.tags = []\n\n        if self.instance:\n            self.tags = self.instance.get('tags', [])\n            # preserve backwards compatible default timeouts\n            if self.instance.get('timeout') is None:\n                if self.init_config.get('timeout') is None:\n                    self.instance['timeout'] = 2\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, self.http)\n            self.instance_cache[instance_key] = {\"config\": config, \"profiler\": profiler}\n\n        if not config.ignore_registry:\n            self._process_registry(config)\n\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        tags = self.tags + [\"source:{0}\".format(source)]\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(\"registry.unprocessed_bytes\", unprocessed_bytes, tags=tags)\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        tags = self.tags + [\"stats_endpoint:{0}\".format(config.stats_endpoint)]\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            self.service_check(\n                self.SERVICE_CHECK_NAME,\n                AgentCheck.CRITICAL,\n                message=\"Error {0} when hitting {1}\".format(ex, config.stats_endpoint),\n            )\n            return\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.__NAMESPACE__) and normalize_metrics:\n                    method(name, value, tags)\n                else:\n                    method(name, value, tags, raw=True)\n        self.service_check(self.SERVICE_CHECK_NAME, AgentCheck.OK, tags=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 os\nimport re\nimport sre_constants\n\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, http):\n        self._config = config\n        self._http = http\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 = self._http.get(self._config.stats_endpoint)\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        self._ignore_registry = instance.get(\"ignore_registry\", False)\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    @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 ignore_registry(self):\n        return self._ignore_registry\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    SERVICE_CHECK_NAME = 'can_connect'\n\n    __NAMESPACE__ = \"filebeat\"\n\n    def __init__(self, *args, **kwargs):\n        AgentCheck.__init__(self, *args, **kwargs)\n        self.instance_cache = {}\n        self.tags = []\n\n        if self.instance:\n            self.tags = self.instance.get('tags', [])\n            # preserve backwards compatible default timeouts\n            if self.instance.get('timeout') is None:\n                if self.init_config.get('timeout') is None:\n                    self.instance['timeout'] = 2\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, self.http)\n            self.instance_cache[instance_key] = {\"config\": config, \"profiler\": profiler}\n\n        if not config.ignore_registry:\n            self._process_registry(config)\n\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\n        except ValueError:\n            contents = []\n            with open(registry_file_path) as registry_file:\n                for line in registry_file:\n                    content = json.loads(line)\n                    if 'op' not in content.keys():\n                        contents.append(content['v'])\n            return contents\n\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        tags = self.tags + [\"source:{0}\".format(source)]\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(\"registry.unprocessed_bytes\", unprocessed_bytes, tags=tags)\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        tags = self.tags + [\"stats_endpoint:{0}\".format(config.stats_endpoint)]\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            self.service_check(\n                self.SERVICE_CHECK_NAME,\n                AgentCheck.CRITICAL,\n                message=\"Error {0} when hitting {1}\".format(ex, config.stats_endpoint),\n            )\n            return\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.__NAMESPACE__) and normalize_metrics:\n                    method(name, value, tags)\n                else:\n                    method(name, value, tags, raw=True)\n        self.service_check(self.SERVICE_CHECK_NAME, AgentCheck.OK, tags=tags)\n", "path": "filebeat/datadog_checks/filebeat/filebeat.py"}]}