[monitoring] Adding influxDB 2.x version support #274

.github/workflows/ci.yml

@@ -62,7 +62,7 @@ jobs:
           pip install -U pip wheel setuptools
 
       - name: Install npm dependencies
-        run: sudo npm install -g install jshint stylelint
+        run: sudo npm install -g jshint stylelint
 
       - name: Start InfluxDB container
         run: docker-compose up -d influxdb

Dockerfile

@@ -1,13 +1,16 @@
 FROM python:3.9.19-slim-bullseye
 
+# Install system dependencies
 RUN apt update && \
     apt install --yes zlib1g-dev libjpeg-dev gdal-bin libproj-dev \
     libgeos-dev libspatialite-dev libsqlite3-mod-spatialite \
     sqlite3 libsqlite3-dev openssl libssl-dev fping && \
     rm -rf /var/lib/apt/lists/* /root/.cache/pip/* /tmp/*
 
+# Upgrade pip and install Python dependencies
 RUN pip install -U pip setuptools wheel
 
+# Copy and install project dependencies
 COPY requirements-test.txt requirements.txt /opt/openwisp/
 RUN pip install -r /opt/openwisp/requirements.txt && \
     pip install -r /opt/openwisp/requirements-test.txt && \
@@ -17,9 +20,11 @@ ADD . /opt/openwisp
 RUN pip install -U /opt/openwisp && \
     rm -rf /var/lib/apt/lists/* /root/.cache/pip/* /tmp/*
 WORKDIR /opt/openwisp/tests/
+# Set environment variables
 ENV NAME=openwisp-monitoring \
     PYTHONBUFFERED=1 \
     INFLUXDB_HOST=influxdb \
     REDIS_HOST=redis
 CMD ["sh", "docker-entrypoint.sh"]
 EXPOSE 8000
+# Command to run the application

README.rst

@@ -326,20 +326,23 @@ Follow the setup instructions of `openwisp-controller
 
     # Make sure you change them in production
     # You can select one of the backends located in openwisp_monitoring.db.backends
-    TIMESERIES_DATABASE = {
+    INFLUXDB_1x_DATABASE = {
         'BACKEND': 'openwisp_monitoring.db.backends.influxdb',
         'USER': 'openwisp',
         'PASSWORD': 'openwisp',
         'NAME': 'openwisp2',
-        'HOST': 'localhost',
+        'HOST': 'influxdb',
         'PORT': '8086',
-        'OPTIONS': {
-            # Specify additional options to be used while initializing
-            # database connection.
-            # Note: These options may differ based on the backend used.
-            'udp_writes': True,
-            'udp_port': 8089,
-        }
+        'OPTIONS': {'udp_writes': False, 'udp_port': 8089},
+    }   
+
+    INFLUXDB_2x_DATABASE = {
+        'BACKEND': 'openwisp_monitoring.db.backends.influxdb2',
+        'TOKEN': 'my-super-secret-auth-token',
+        'ORG': 'openwisp', 
+        'BUCKET': 'openwisp2',
+        'HOST': 'influxdb2',
+        'PORT': '9999',
     }
 
 ``urls.py``:
@@ -1413,56 +1416,109 @@ Settings
 | **default**: | see below |
 +--------------+-----------+
 
+Timeseries Database Configuration
+---------------------------------
+
+The ``TIMESERIES_DATABASE`` setting allows configuring the timeseries 
+database backend used by OpenWISP Monitoring. The configuration supports 
+both InfluxDB 1.x and 2.x versions.
+
+Configuration for InfluxDB 1.x
+------------------------------
+
 .. code-block:: python
 
-    TIMESERIES_DATABASE = {
+    INFLUXDB_1x_DATABASE = {
         'BACKEND': 'openwisp_monitoring.db.backends.influxdb',
         'USER': 'openwisp',
         'PASSWORD': 'openwisp',
         'NAME': 'openwisp2',
-        'HOST': 'localhost',
+        'HOST': 'influxdb',
         'PORT': '8086',
-        'OPTIONS': {
-            'udp_writes': False,
-            'udp_port': 8089,
-        }
+        'OPTIONS': {'udp_writes': False, 'udp_port': 8089},
     }
 
-The following table describes all keys available in ``TIMESERIES_DATABASE``
-setting:
-
-+---------------+--------------------------------------------------------------------------------------+
-| **Key**       | ``Description``                                                                      |
-+---------------+--------------------------------------------------------------------------------------+
-| ``BACKEND``   | The timeseries database backend to use. You can select one of the backends           |
-|               | located in ``openwisp_monitoring.db.backends``                                       |
-+---------------+--------------------------------------------------------------------------------------+
-| ``USER``      | User for logging into the timeseries database                                        |
-+---------------+--------------------------------------------------------------------------------------+
-| ``PASSWORD``  | Password of the timeseries database user                                             |
-+---------------+--------------------------------------------------------------------------------------+
-| ``NAME``      | Name of the timeseries database                                                      |
-+---------------+--------------------------------------------------------------------------------------+
-| ``HOST``      | IP address/hostname of machine where the timeseries database is running              |
-+---------------+--------------------------------------------------------------------------------------+
-| ``PORT``      | Port for connecting to the timeseries database                                       |
-+---------------+--------------------------------------------------------------------------------------+
-| ``OPTIONS``   | These settings depends on the timeseries backend:                                    |
-|               |                                                                                      |
-|               | +-----------------+----------------------------------------------------------------+ |
-|               | | ``udp_writes``  | Whether to use UDP for writing data to the timeseries database | |
-|               | +-----------------+----------------------------------------------------------------+ |
-|               | | ``udp_port``    | Timeseries database port for writing data using UDP            | |
-|               | +-----------------+----------------------------------------------------------------+ |
-+---------------+--------------------------------------------------------------------------------------+
-
-**Note:** UDP packets can have a maximum size of 64KB. When using UDP for writing timeseries
-data, if the size of the data exceeds 64KB, TCP mode will be used instead.
-
-**Note:** If you want to use the ``openwisp_monitoring.db.backends.influxdb`` backend
-with UDP writes enabled, then you need to enable two different ports for UDP
-(each for different retention policy) in your InfluxDB configuration. The UDP configuration
-section of your InfluxDB should look similar to the following:
+Configuration for InfluxDB 2.x
+------------------------------
+
+.. code-block:: python
+
+    INFLUXDB_2x_DATABASE = {
+        'BACKEND': 'openwisp_monitoring.db.backends.influxdb2',
+        'TOKEN': 'my-super-secret-auth-token',
+        'ORG': 'openwisp',
+        'BUCKET': 'openwisp2',
+        'HOST': 'influxdb2',
+        'PORT': '9999',
+    }
+
+Dynamic Configuration Based on Environment
+------------------------------------------
+
+You can dynamically switch between InfluxDB 1.x and 2.x configurations 
+using environment variables:
+
+.. code-block:: python
+
+    import os
+
+    if os.environ.get('USE_INFLUXDB2', 'False') == 'True':
+        TIMESERIES_DATABASE = INFLUXDB_2x_DATABASE
+    else:
+        TIMESERIES_DATABASE = INFLUXDB_1x_DATABASE
+
+    if TESTING:
+        if os.environ.get('TIMESERIES_UDP', False):
+            TIMESERIES_DATABASE['OPTIONS'] = {'udp_writes': True, 'udp_port': 8091}
+
+Explanation of Settings
+-----------------------
+
++---------------+---------------------------------------------------------------+
+| **Key**       | **Description**                                               |
++-------------------------------------------------------------------------------+
+| ``BACKEND``   | The timeseries database backend to use. You can select one    |
+|               | of the backends located in ``openwisp_monitoring.db.backends``|
++---------------+---------------------------------------------------------------+
+| ``USER``      | User for logging into the timeseries database (only for       |
+|               | InfluxDB 1.x)                                                 |
++---------------+---------------------------------------------------------------+
+| ``PASSWORD``  | Password of the timeseries database user (only for InfluxDB   |
+|               | 1.x)                                                          |
++---------------+---------------------------------------------------------------+
+| ``NAME``      | Name of the timeseries database (only for InfluxDB 1.x)       |
++---------------+---------------------------------------------------------------+
+| ``TOKEN``     | Authentication token for InfluxDB 2.x                         |
++---------------+---------------------------------------------------------------+
+| ``ORG``       | Organization name for InfluxDB 2.x                            |
++---------------+---------------------------------------------------------------+
+| ``BUCKET``    | Bucket name for InfluxDB 2.x                                  |
++---------------+---------------------------------------------------------------+
+| ``HOST``      | IP address/hostname of machine where the timeseries           |
+|               | database is running                                           |
++---------------+---------------------------------------------------------------+
+| ``PORT``      | Port for connecting to the timeseries database                |
++---------------+---------------------------------------------------------------+
+| ``OPTIONS``   | Additional options for the timeseries backend                 |
+|               |                                                               |
+|               | +-----------------+-----------------------------------------+ |
+|               | | ``udp_writes``  | Whether to use UDP for writing data     | |
+|               | |                 | to the timeseries database              | |
+|               | +-----------------+-----------------------------------------+ |
+|               | | ``udp_port``    | Timeseries database port for writing    | |
+|               | |                 | data using UDP                          | |
+|               | +-----------------+-----------------------------------------+ |
++---------------+---------------------------------------------------------------+
+
+UDP Configuration for InfluxDB 1.x
+----------------------------------
+
+If you want to use the ``openwisp_monitoring.db.backends.influxdb`` backend 
+with UDP writes enabled, you need to enable two different ports for UDP 
+(each for a different retention policy) in your InfluxDB configuration.
+
+Here is an example of the UDP configuration section in your InfluxDB 
+configuration file:
 
 .. code-block:: text
 
@@ -1479,6 +1535,13 @@ section of your InfluxDB should look similar to the following:
     database = "openwisp2"
     retention-policy = 'short'
 
+**Note:** UDP packets can have a maximum size of 64KB. When using UDP for 
+writing timeseries data, if the size of the data exceeds 64KB, TCP mode 
+will be used instead.
+
+Deploying with Ansible
+----------------------
+
 If you are using `ansible-openwisp2 <https://github.com/openwisp/ansible-openwisp2>`_
 for deploying OpenWISP, you can set the ``influxdb_udp_mode`` ansible variable to ``true``
 in your playbook, this will make the ansible role automatically configure the InfluxDB UDP listeners.

docker-compose.yml

@@ -28,6 +28,22 @@ services:
       INFLUXDB_USER: openwisp
       INFLUXDB_USER_PASSWORD: openwisp
 
+  influxdb2:
+    image: influxdb:2.0
+    container_name: influxdb2
+    ports:
+      # Map the 9086 port on host machine to 8086 in container
+      - "9086:8086"
+    environment:
+      DOCKER_INFLUXDB_INIT_MODE: setup
+      DOCKER_INFLUXDB_INIT_USERNAME: myuser
+      DOCKER_INFLUXDB_INIT_PASSWORD: mypassword
+      DOCKER_INFLUXDB_INIT_ORG: myorg
+      DOCKER_INFLUXDB_INIT_BUCKET: mybucket
+      DOCKER_INFLUXDB_INIT_RETENTION: 1w
+    volumes:
+      - influxdb-storage:/var/lib/influxdb2
+
   redis:
     image: redis:5.0-alpine
     ports:
@@ -36,3 +52,4 @@ services:
 
 volumes:
   influxdb-data: {}
+  influxdb-storage:

openwisp_monitoring/db/__init__.py

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

openwisp_monitoring/db/backends/__init__.py

@@ -30,11 +30,16 @@ def load_backend_module(backend_name=TIMESERIES_DB['BACKEND'], module=None):
     """
     try:
         assert 'BACKEND' in TIMESERIES_DB, 'BACKEND'
-        assert 'USER' in TIMESERIES_DB, 'USER'
-        assert 'PASSWORD' in TIMESERIES_DB, 'PASSWORD'
-        assert 'NAME' in TIMESERIES_DB, 'NAME'
-        assert 'HOST' in TIMESERIES_DB, 'HOST'
-        assert 'PORT' in TIMESERIES_DB, 'PORT'
+        if 'BACKEND' in TIMESERIES_DB and '2' in TIMESERIES_DB['BACKEND']:
+            # InfluxDB 2.x specific checks
+            assert 'TOKEN' in TIMESERIES_DB, 'TOKEN'
+            assert 'ORG' in TIMESERIES_DB, 'ORG'
+            assert 'BUCKET' in TIMESERIES_DB, 'BUCKET'
+        else:
+            # InfluxDB 1.x specific checks
+            assert 'USER' in TIMESERIES_DB, 'USER'
+            assert 'PASSWORD' in TIMESERIES_DB, 'PASSWORD'
+            assert 'NAME' in TIMESERIES_DB, 'NAME'
         if module:
             return import_module(f'{backend_name}.{module}')
         else:
@@ -48,7 +53,7 @@ def load_backend_module(backend_name=TIMESERIES_DB['BACKEND'], module=None):
     except ImportError as e:
         # The database backend wasn't found. Display a helpful error message
         # listing all built-in database backends.
-        builtin_backends = ['influxdb']
+        builtin_backends = ['influxdb', 'influxdb2']
         if backend_name not in [
             f'openwisp_monitoring.db.backends.{b}' for b in builtin_backends
         ]:

openwisp_monitoring/db/backends/influxdb/client.py

@@ -56,7 +56,6 @@ class DatabaseClient(object):
     backend_name = 'influxdb'
 
     def __init__(self, db_name=None):
-        self._db = None
         self.db_name = db_name or TIMESERIES_DB['NAME']
         self.client_error = InfluxDBClientError
 
@@ -255,7 +254,7 @@ def read(self, key, fields, tags, **kwargs):
             q = f'{q} LIMIT {limit}'
         return list(self.query(q, precision='s').get_points())
 
-    def get_list_query(self, query, precision='s'):
+    def get_list_query(self, query, precision='s', **kwargs):
         result = self.query(query, precision=precision)
         if not len(result.keys()) or result.keys()[0][1] is None:
             return list(result.get_points())
@@ -426,16 +425,23 @@ def __transform_field(self, field, function, operation=None):
 
     def _get_top_fields(
         self,
+        default_query,
         query,
         params,
         chart_type,
         group_map,
         number,
         time,
         timezone=settings.TIME_ZONE,
+        get_fields=True,
     ):
+        """
+        Returns top fields if ``get_fields`` set to ``True`` (default)
+        else it returns points containing the top fields.
+        """
+        q = default_query.replace('{field_name}', '{fields}')
         q = self.get_query(
-            query=query,
+            query=q,
             params=params,
             chart_type=chart_type,
             group_map=group_map,
@@ -444,7 +450,7 @@ def _get_top_fields(
             time=time,
             timezone=timezone,
         )
-        res = list(self.query(q, precision='s').get_points())
+        res = self.get_list_query(q)
         if not res:
             return []
         res = res[0]
@@ -454,4 +460,31 @@ def _get_top_fields(
         keys = list(sorted_dict.keys())
         keys.reverse()
         top = keys[0:number]
-        return [item.replace('sum_', '') for item in top]
+        top_fields = [item.replace('sum_', '') for item in top]
+        if get_fields:
+            return top_fields
+        query = self.get_query(
+            query=query,
+            params=params,
+            chart_type=chart_type,
+            group_map=group_map,
+            summary=True,
+            fields=top_fields,
+            time=time,
+            timezone=timezone,
+        )
+        return self.get_list_query(query)
+
+    def default_chart_query(self, tags):
+        q = "SELECT {field_name} FROM {key} WHERE time >= '{time}'"
+        if tags:
+            q += " AND content_type = '{content_type}' AND object_id = '{object_id}'"
+        return q
+
+    def _device_data(self, key, tags, rp, **kwargs):
+        """ returns last snapshot of ``device_data`` """
+        query = (
+            f"SELECT data FROM {rp}.{key} WHERE pk = '{tags['pk']}' "
+            "ORDER BY time DESC LIMIT 1"
+        )
+        return self.get_list_query(query, precision=None)