[timeseries] Add top fields support

Author: nepython

README.rst

@@ -24,7 +24,7 @@ Available Features
 
 * Collects and displays device status information like uptime, RAM status, CPU load averages,
   Interface addresses, WiFi interface status and associated clients, Neighbors information, DHCP Leases, Disk/Flash status
-* Collection of monitoring information in a timeseries database (currently only influxdb is supported)
+* Collection of monitoring information in a timeseries database (`InfluxDB <https://www.influxdata.com/>`_ and `Elasticsearch <https://www.elastic.co/elasticsearch/>`_ are currently supported)
 * Monitoring charts for uptime, packet loss, round trip time (latency), associated wifi clients, interface traffic,
   RAM usage, CPU load, flash/disk usage
 * Charts can be viewed at resolutions of 1 day, 3 days, a week, a month and a year
@@ -46,6 +46,8 @@ beforehand.
 In case you prefer not to use Docker you can `install InfluxDB <https://docs.influxdata.com/influxdb/v1.8/introduction/install/>`_
 and Redis from your repositories, but keep in mind that the version packaged by your distribution may be different.
 
+If you wish to use ``Elasticsearch`` for storing and retrieving timeseries data then `install Elasticsearch <https://www.elastic.co/guide/en/elasticsearch/reference/current/install-elasticsearch.html>`_.
+
 Install spatialite and sqlite:
 
 .. code-block:: shell
@@ -106,6 +108,19 @@ Follow the setup instructions of `openwisp-controller
         'PORT': '8086',
     }
 
+In case, you wish to use ``Elasticsearch`` for timeseries data storage and retrieval,
+make use i=of the following settings
+
+.. code-block:: python
+    TIMESERIES_DATABASE = {
+        'BACKEND': 'openwisp_monitoring.db.backends.elasticsearch',
+        'USER': 'openwisp',
+        'PASSWORD': 'openwisp',
+        'NAME': 'openwisp2',
+        'HOST': 'localhost',
+        'PORT': '9200',
+    }
+
 ``urls.py``:
 
 .. code-block:: python
@@ -231,6 +246,9 @@ This data is only used to assess the recent status of devices, keeping
 it for a long time would not add much benefit and would cost a lot more
 in terms of disk space.
 
+**Note**: In case you use ``Elasticsearch`` then time shall be taken as integral multiple of a day.
+That means the time ``36h0m0s`` shall be interpreted as ``24h0m0s`` (integral multiple of a day).
+
 ``OPENWISP_MONITORING_AUTO_PING``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -360,18 +378,30 @@ MB (megabytes) instead of GB (Gigabytes) you can use:
                     "SUM(rx_bytes) / 1000000 AS download FROM {key} "
                     "WHERE time >= '{time}' AND content_type = '{content_type}' "
                     "AND object_id = '{object_id}' GROUP BY time(1d)"
-                )
+                ),
+                'elasticsearch': _make_query({
+                    'upload': {'sum': {'field': 'points.fields.tx_bytes'}},
+                    'download': {'avg': {'field': 'points.fields.rx_bytes'}},
+                })
             },
         }
     }
 
+    # This needs to be declared separately but only for elasticsearch
+    OPENWISP_MONITORING_ADDITIONAL_CHARTS_OPERATIONS = {
+        'upload': {'operator': '/', 'value': 1000000},
+        'download': {'operator': '/', 'value': 1000000},
+    }
+
 Or if you want to define a new chart configuration, which you can then
 call in your custom code (eg: a custom check class), you can do so as follows:
 
 .. code-block:: python
 
     from django.utils.translation import gettext_lazy as _
 
+    from openwisp_monitoring.db.backends.elasticsearch import _make_query
+
     OPENWISP_MONITORING_CHARTS = {
         'ram': {
             'type': 'line',
@@ -385,7 +415,12 @@ call in your custom code (eg: a custom check class), you can do so as follows:
                     "MEAN(buffered) AS buffered FROM {key} WHERE time >= '{time}' AND "
                     "content_type = '{content_type}' AND object_id = '{object_id}' "
                     "GROUP BY time(1d)"
-                )
+                ),
+                'elasticsearch': _make_query({
+                    'total': {'avg': {'field': 'points.fields.total'}},
+                    'free': {'avg': {'field': 'points.fields.free'}},
+                    'buffered': {'avg': {'field': 'points.fields.buffered'}},
+                })
             },
         }
     }

openwisp_monitoring/db/__init__.py

@@ -1,6 +1,5 @@
 from .backends import timeseries_db
 
 chart_query = timeseries_db.queries.chart_query
-device_data_query = timeseries_db.queries.device_data_query
 
-__all__ = ['timeseries_db', 'chart_query', 'device_data_query']
+__all__ = ['timeseries_db', 'chart_query']

openwisp_monitoring/db/backends/elasticsearch/__init__.py

@@ -0,0 +1,3 @@
+from .queries import _make_query
+
+__all__ = ['_make_query']

openwisp_monitoring/db/backends/elasticsearch/client.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from collections import Counter
 from copy import deepcopy
 from datetime import datetime, timedelta
 
@@ -70,7 +71,9 @@ def __init__(self, db_name='metric'):
 
     def create_database(self):
         """ creates connection to elasticsearch """
-        connections.create_connection(hosts=[f"{TIMESERIES_DB['HOST']}:{TIMESERIES_DB['PORT']}"])
+        connections.create_connection(
+            hosts=[f"{TIMESERIES_DB['HOST']}:{TIMESERIES_DB['PORT']}"]
+        )
         db = self.get_db
         # Skip if support for Index Lifecycle Management is disabled or no privileges
         self.ilm_enabled = db.ilm.start()['acknowledged']
@@ -122,15 +125,16 @@ def create_or_alter_retention_policy(self, name, duration=None):
             ilm.put_lifecycle(policy=name, body=policy)
 
     def query(self, query, precision=None):
-        index = query.pop('key')
-        return Search(using=self.get_db, index=index).from_dict(query).execute().to_dict()
+        if 'summary' in query:
+            query.pop('summary')
+        return Search(using=self.get_db).from_dict(query).execute().to_dict()
 
     def write(self, name, values, **kwargs):
         rp = kwargs.get('retention_policy')
         tags = kwargs.get('tags')
         timestamp = kwargs.get('timestamp')
-        metric_id = find_metric(self.get_db, name, tags, rp, add=True)
-        metric_index = MetricIndex.get(metric_id, index=name, using=self.get_db)
+        metric_id, index = find_metric(self.get_db, name, tags, rp, add=True)
+        metric_index = MetricIndex.get(metric_id, index=index, using=self.get_db)
         point = Point(fields=values, time=timestamp or datetime.now())
         metric_index.points.append(point)
         metric_index.save()
@@ -140,10 +144,11 @@ def read(self, key, fields, tags, limit=1, order='-time', **kwargs):
         time_format = kwargs.get('time_format')
         # TODO: It will be of the form 'now() - <int>s'
         # since = kwargs.get('since')
-        metric_id = find_metric(self.get_db, key, tags)
-        if not metric_id:
+        try:
+            metric_id, index = find_metric(self.get_db, key, tags)
+        except TypeError:
             return []
-        metric_index = MetricIndex.get(index=key, id=metric_id, using=self.get_db)
+        metric_index = MetricIndex.get(index=index, id=metric_id, using=self.get_db)
         if order == 'time':
             points = list(metric_index.points[0:limit])
         elif order == '-time':
@@ -195,26 +200,31 @@ def _format_time(self, obj, time_format=None):
 
     def get_list_query(self, query, precision='s'):
         response = self.query(query, precision)
-        points = response['aggregations']['GroupByTime']['buckets']
-        list_points = self._fill_points(
-            query, [self._format(point) for point in points]
-        )
+        try:
+            points = response['aggregations']['GroupByTime']['set_range']['time'][
+                'buckets'
+            ]
+            list_points = self._fill_points(
+                query, [self._format(point) for point in points],
+            )
+        except KeyError:
+            return []
         return list_points
 
     def _fill_points(self, query, points):
-        _range = next(
-            (item for item in query['query']['bool']['must'] if item.get('range')), None
-        )
+        _range = query['aggs']['GroupByTime']['nested']['aggs']['set_range']
         if not _range or not points:
             return points
-        days = int(_range['range']['points.time']['from'][4:-3])
+        days = int(_range['filter']['range']['points.time']['from'][4:-3])
+        interval = _range['aggs']['time']['date_histogram']['fixed_interval']
+        # Check if summary query
+        if f'{days}d' == interval:
+            return points
+        interval_dict = {'10m': 600, '20m': 1200, '1h': 3600, '24h': 86400}
+        interval = interval_dict[interval]
         start_time = datetime.now()
         end_time = start_time - timedelta(days=days)  # include today
         dummy_point = deepcopy(points[0])
-        if len(points) > 2:
-            interval = points[0]['time'] - points[1]['time']
-        else:
-            interval = 600
         start_ts = points[0]['time'] + interval
         end_ts = points[-1]['time'] - interval
         for field in dummy_point.keys():
@@ -223,7 +233,7 @@ def _fill_points(self, query, points):
             dummy_point['time'] = start_ts
             points.insert(0, deepcopy(dummy_point))
             start_ts += interval
-        # TODO: This needs to be fixed and shouldn't be required since intervals are set
+        # TODO: Why is this required since intervals are set?
         while points[-1]['time'] < end_time.timestamp():
             points.pop(-1)
         while end_ts > end_time.timestamp():
@@ -238,8 +248,8 @@ def delete_metric_data(self, key=None, tags=None):
         deletes all metrics if neither provided
         """
         if key and tags:
-            metric_id = find_metric(self.get_db, key, tags)
-            self.get_db.delete(index=key, id=metric_id)
+            metric_id, index = find_metric(self.get_db, key, tags)
+            self.get_db.delete(index=index, id=metric_id)
         elif key:
             self.get_db.indices.delete(index=key, ignore=[400, 404])
         else:
@@ -252,14 +262,19 @@ def validate_query(self, query):
             query = json.loads(query)
         # Elasticsearch currently supports validation of only query section,
         # aggs, size, _source etc. are not supported
-        valid_check = self.get_db.indices.validate_query(body={'query': query['query']}, explain=True)
+        valid_check = self.get_db.indices.validate_query(
+            body={'query': query['query']}, explain=True
+        )
         # Show a helpful message for failure
         if not valid_check['valid']:
-            raise ValidationError(valid_check['explanations'])
+            raise ValidationError(valid_check['error'])
         return self._is_aggregate(query)
 
+    # TODO: This is not covering everything
     def _is_aggregate(self, q):
-        agg_dict = q['aggs']['GroupByTime']['aggs'].values()
+        agg_dict = q['aggs']['GroupByTime']['nested']['aggs']['set_range']['aggs'][
+            'time'
+        ]['aggs']['nest']['nested']['aggs'].values()
         agg = []
         for item in agg_dict:
             agg.append(next(iter(item)))
@@ -276,53 +291,88 @@ def get_query(
         query=None,
         timezone=settings.TIME_ZONE,
     ):
-        query['key'] = params.pop('key')
         query = json.dumps(query)
         for k, v in params.items():
             query = query.replace('{' + k + '}', v)
         query = self._group_by(query, time, chart_type, group_map, strip=summary)
         query = json.loads(query)
-        if summary:
-            _range = next(
-                (item for item in query['query']['bool']['must'] if item.get('range')),
-                None,
+        set_range = query['aggs']['GroupByTime']['nested']['aggs']['set_range']['aggs']['time']
+        if fields:
+            aggregate_dict = set_range['aggs']['nest']['nested']['aggs']
+            agg = deepcopy(aggregate_dict).popitem()[1].popitem()[0]
+            aggregate_dict.update(
+                {
+                    f'{field}': {agg: {'field': f'points.fields.{field}'}}
+                    for field in fields
+                }
             )
-            if _range:
-                query['query']['bool']['must'].remove(_range)
-        query['aggs']['GroupByTime']['date_histogram']['time_zone'] = timezone
+        try:
+            set_range['date_histogram']['time_zone'] = timezone
+        except KeyError:
+            pass
         return query
 
     def _group_by(self, query, time, chart_type, group_map, strip=False):
-        if not self.validate_query(query):
-            return query
+        query = query.replace('1d/d', f'{time}/d')
         if not strip and not chart_type == 'histogram':
             value = group_map[time]
-            query = query.replace('1d/d', f'{time}/d')
             query = query.replace('10m', value)
         if strip:
             query = query.replace('10m', time)
         return query
 
-    # TODO:
     def _get_top_fields(
         self,
-        query,
         params,
         chart_type,
         group_map,
         number,
         time,
+        query=None,
         timezone=settings.TIME_ZONE,
+        get_fields=True,
+        **kwargs,
     ):
-        pass
+        """
+        Returns top fields if ``get_fields`` set to ``True`` (default)
+        else it returns points containing the top fields.
+        """
+        response = self.get_db.indices.get_mapping(index=params['key'])
+        fields = [
+            k
+            for k, v in list(response.values())[0]['mappings']['properties']['points'][
+                'properties'
+            ]['fields']['properties'].items()
+        ]
+        query = self.get_query(
+            chart_type,
+            params,
+            time,
+            group_map,
+            summary=True,
+            fields=fields,
+            query=query,
+            timezone=timezone,
+        )
+        point = self.get_list_query(query)[0]
+        time = point.pop('time')
+        point = Counter(point).most_common(number)
+        if get_fields:
+            return [k for k, v in point]
+        points = [{'time': time}]
+        for k, v in point:
+            points[0].update({k: v})
+        return points
 
     def _format(self, point):
         pt = {}
         # Convert time from milliseconds -> seconds precision
         pt['time'] = int(point['key'] / 1000)
         for key, value in point.items():
             if isinstance(value, dict):
-                pt[key] = self._transform_field(key, value['value'])
+                for k, v in value.items():
+                    if isinstance(v, dict):
+                        pt[k] = self._transform_field(k, v['value'])
         return pt
 
     def _transform_field(self, field, value):
@@ -338,12 +388,10 @@ def _transform_field(self, field, value):
     def default_chart_query(self, tags):
         q = deepcopy(default_chart_query)
         if not tags:
-            q['query']['bool']['must'].pop(0)
-            q['query']['bool']['must'].pop(1)
+            q['query']['nested']['query']['bool']['must'].pop(0)
+            q['query']['nested']['query']['bool']['must'].pop(1)
         return q
 
-
-# TODO:
-# Fix Average - currently it's computing average over all fields!
-# Time Interval - fix range
-# Device query
+    def _device_data(self, key, tags, fields, **kwargs):
+        """ returns last snapshot of ``device_data`` """
+        return self.read(key=key, fields=fields, tags=tags, time_format='isoformat',)