We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Filebeat instance cache is broken (which breaks increment metrics)
**Output of the [info page](https://docs.datadoghq.com/agent/faq/agent-commands/#agent-status-and-information)**

```text
====================
Collector (v 5.30.1)
====================

  Status date: 2019-01-08 06:09:58 (10s ago)
  Pid: 4468
  Platform: Linux-3.13.0-163-generic-x86_64-with-Ubuntu-14.04-trusty
  Python Version: 2.7.15, 64bit
  Logs: <stderr>, /var/log/datadog/collector.log
<snip>
    filebeat (custom)
    -----------------
      - instance #0 [OK]
      - Collected 1 metric, 0 events & 0 service checks
<snip>
```

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

**Steps to reproduce the issue:**
1. Install and configure filebeat check with:
```yaml
init_config:

instances:
  - registry_file_path: /var/lib/filebeat/registry
    stats_endpoint: http://localhost:5066/stats
```

**Describe the results you received:**
Metrics in `GAUGE_METRIC_NAMES` are reported to Datadog, but none of the metrics in `INCREMENT_METRIC_NAMES` are.

**Describe the results you expected:**
All listed metrics in the above variables and present in `http://localhost:5066/stats` are reported.

**Additional information you deem important (e.g. issue happens only occasionally):**
The issue is here: https://github.com/DataDog/integrations-extras/blob/2eff6e2dd2b123214cd562b93a9b45753e27a959/filebeat/check.py#L211-L218

Line 213 checks for `instance_key` in `self.instance_cache`, but that key is never actually put in that dictionary. The only keys in that dictionary are `'config'` and `'profiler'`, added on lines 217 and 218.

The result of this is that the `FilebeatCheckInstanceConfig` and `FilebeatCheckHttpProfiler` are re-created every check, which means that the `_previous_increment_values` property on FilebeatCheckHttpProfiler is always empty, which means no increment metrics are ever reported.

This appears to have been introduced in #250.

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

{"golden_diff": "diff --git a/filebeat/check.py b/filebeat/check.py\n--- a/filebeat/check.py\n+++ b/filebeat/check.py\n@@ -211,11 +211,12 @@\n     def check(self, instance):\n         instance_key = hash_mutable(instance)\n         if instance_key in self.instance_cache:\n-            config = self.instance_cache['config']\n-            profiler = self.instance_cache['profiler']\n+            config = self.instance_cache[instance_key]['config']\n+            profiler = self.instance_cache[instance_key]['profiler']\n         else:\n-            self.instance_cache['config'] = config = FilebeatCheckInstanceConfig(instance)\n-            self.instance_cache['profiler'] = profiler = FilebeatCheckHttpProfiler(config)\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)\n", "issue": "Filebeat instance cache is broken (which breaks increment metrics)\n**Output of the [info page](https://docs.datadoghq.com/agent/faq/agent-commands/#agent-status-and-information)**\r\n\r\n```text\r\n====================\r\nCollector (v 5.30.1)\r\n====================\r\n\r\n  Status date: 2019-01-08 06:09:58 (10s ago)\r\n  Pid: 4468\r\n  Platform: Linux-3.13.0-163-generic-x86_64-with-Ubuntu-14.04-trusty\r\n  Python Version: 2.7.15, 64bit\r\n  Logs: <stderr>, /var/log/datadog/collector.log\r\n<snip>\r\n    filebeat (custom)\r\n    -----------------\r\n      - instance #0 [OK]\r\n      - Collected 1 metric, 0 events & 0 service checks\r\n<snip>\r\n```\r\n\r\n**Additional environment details (Operating System, Cloud provider, etc):**\r\nFilebeat version 6.5.1\r\n\r\n**Steps to reproduce the issue:**\r\n1. Install and configure filebeat check with:\r\n```yaml\r\ninit_config:\r\n\r\ninstances:\r\n  - registry_file_path: /var/lib/filebeat/registry\r\n    stats_endpoint: http://localhost:5066/stats\r\n```\r\n\r\n**Describe the results you received:**\r\nMetrics in `GAUGE_METRIC_NAMES` are reported to Datadog, but none of the metrics in `INCREMENT_METRIC_NAMES` are.\r\n\r\n**Describe the results you expected:**\r\nAll listed metrics in the above variables and present in `http://localhost:5066/stats` are reported.\r\n\r\n**Additional information you deem important (e.g. issue happens only occasionally):**\r\nThe issue is here: https://github.com/DataDog/integrations-extras/blob/2eff6e2dd2b123214cd562b93a9b45753e27a959/filebeat/check.py#L211-L218\r\n\r\nLine 213 checks for `instance_key` in `self.instance_cache`, but that key is never actually put in that dictionary. The only keys in that dictionary are `'config'` and `'profiler'`, added on lines 217 and 218.\r\n\r\nThe result of this is that the `FilebeatCheckInstanceConfig` and `FilebeatCheckHttpProfiler` are re-created every check, which means that the `_previous_increment_values` property on FilebeatCheckHttpProfiler is always empty, which means no increment metrics are ever reported.\r\n\r\nThis appears to have been introduced in #250.\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 numbers\nimport os\nimport re\nimport sre_constants\n\n# 3rd party\nimport requests\nimport simplejson\n\n# project\nfrom datadog_checks.checks import AgentCheck\nfrom datadog_checks.utils.containers import hash_mutable\n\n\nEVENT_TYPE = SOURCE_TYPE_NAME = 'filebeat'\n\n\nclass FilebeatCheckHttpProfiler(object):\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.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.dropped',\n        'libbeat.output.events.failed',\n        'libbeat.pipeline.events.dropped',\n        'libbeat.pipeline.events.failed',\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 = [\n        'filebeat.harvester.running'\n    ]\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        response = self._make_request()\n\n        return {\n            'increment': self._gather_increment_metrics(response),\n            'gauge': self._gather_gauge_metrics(response)\n        }\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 = {name: response[name] for name in self.INCREMENT_METRIC_NAMES\n                      if self._should_keep_metric(name) and name in response}\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 new_values.iteritems():\n            if name not in self._previous_increment_values \\\n                    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 {name: response[name] for name in self.GAUGE_METRIC_NAMES\n                if self._should_keep_metric(name) and name in response}\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(object):\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        if not isinstance(self._only_metrics, list):\n            raise Exception(\"If given, filebeat's only_metrics must be a list of regexes, got %s\" % (\n                self._only_metrics, ))\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        try:\n            return self._only_metrics_regexes\n        except AttributeError:\n            self._only_metrics_regexes = self._compile_regexes()\n            return self._compiled_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(\n                    'Invalid only_metric regex for filebeat: \"%s\", error: %s' % (regex, ex))\n\n        return compiled_regexes\n\n\nclass FilebeatCheck(AgentCheck):\n\n    def __init__(self, *args, **kwargs):\n        AgentCheck.__init__(self, *args, **kwargs)\n        self.instance_cache = {}\n\n    def check(self, instance):\n        instance_key = hash_mutable(instance)\n        if instance_key in self.instance_cache:\n            config = self.instance_cache['config']\n            profiler = self.instance_cache['profiler']\n        else:\n            self.instance_cache['config'] = config = FilebeatCheckInstanceConfig(instance)\n            self.instance_cache['profiler'] = profiler = FilebeatCheckHttpProfiler(config)\n\n        self._process_registry(config)\n        self._gather_http_profiler_metrics(config, profiler)\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 simplejson.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('You might be interesting in having a look at https://github.com/elastic/beats/pull/6455')\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,\n                           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):\n        try:\n            all_metrics = profiler.gather_metrics()\n        except StandardError 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 all_metrics.iteritems():\n            method = getattr(self, action)\n\n            for name, value in metrics.iteritems():\n                method(name, value, tags)\n", "path": "filebeat/check.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 collections\nimport errno\nimport numbers\nimport os\nimport re\nimport sre_constants\n\n# 3rd party\nimport requests\nimport simplejson\n\n# project\nfrom datadog_checks.checks import AgentCheck\nfrom datadog_checks.utils.containers import hash_mutable\n\n\nEVENT_TYPE = SOURCE_TYPE_NAME = 'filebeat'\n\n\nclass FilebeatCheckHttpProfiler(object):\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.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.dropped',\n        'libbeat.output.events.failed',\n        'libbeat.pipeline.events.dropped',\n        'libbeat.pipeline.events.failed',\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 = [\n        'filebeat.harvester.running'\n    ]\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        response = self._make_request()\n\n        return {\n            'increment': self._gather_increment_metrics(response),\n            'gauge': self._gather_gauge_metrics(response)\n        }\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 = {name: response[name] for name in self.INCREMENT_METRIC_NAMES\n                      if self._should_keep_metric(name) and name in response}\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 new_values.iteritems():\n            if name not in self._previous_increment_values \\\n                    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 {name: response[name] for name in self.GAUGE_METRIC_NAMES\n                if self._should_keep_metric(name) and name in response}\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(object):\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        if not isinstance(self._only_metrics, list):\n            raise Exception(\"If given, filebeat's only_metrics must be a list of regexes, got %s\" % (\n                self._only_metrics, ))\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        try:\n            return self._only_metrics_regexes\n        except AttributeError:\n            self._only_metrics_regexes = self._compile_regexes()\n            return self._compiled_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(\n                    'Invalid only_metric regex for filebeat: \"%s\", error: %s' % (regex, ex))\n\n        return compiled_regexes\n\n\nclass FilebeatCheck(AgentCheck):\n\n    def __init__(self, *args, **kwargs):\n        AgentCheck.__init__(self, *args, **kwargs)\n        self.instance_cache = {}\n\n    def check(self, instance):\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)\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 simplejson.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('You might be interesting in having a look at https://github.com/elastic/beats/pull/6455')\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,\n                           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):\n        try:\n            all_metrics = profiler.gather_metrics()\n        except StandardError 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 all_metrics.iteritems():\n            method = getattr(self, action)\n\n            for name, value in metrics.iteritems():\n                method(name, value, tags)\n", "path": "filebeat/check.py"}]}