[fix] Fixed "migrate_timeseries" command for UDP writing #625

Fixes #626
openwisp · Dec 27, 2024 · abeb3a6 · abeb3a6
1 parent c8aa451
commit abeb3a6
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 7 deletions.
diff --git a/openwisp_monitoring/db/backends/influxdb/client.py b/openwisp_monitoring/db/backends/influxdb/client.py
@@ -148,15 +148,33 @@ def query(self, query, precision=None, **kwargs):
             database=database,
         )
 
-    def _write(self, points, database, retention_policy):
+    def store(self, points, database, retention_policy):
+        """
+        Store data points in the specified database.
+
+        This method writes data points to the specified database. If the size of the data exceeds
+        the limit of a UDP packet, it falls back to using a TCP connection for writing data.
+
+        Args:
+            points (list): The data points to be stored.
+            database (str): The name of the database where the data points will be stored.
+            retention_policy (str): The retention policy to be used for storing the data points.
+
+        Returns:
+            bool: True if the data points were successfully written, False otherwise.
+
+        Raises:
+            TimeseriesWriteException: If there is an error while writing the data points.
+        """
         db = self.dbs['short'] if retention_policy else self.dbs['default']
         # If the size of data exceeds the limit of the UDP packet, then
         # fallback to use TCP connection for writing data.
         lines = make_lines({'points': points})
         if sys.getsizeof(lines) > 65000:
+            # Size exceeds UDP limit, write using TCP.
             db = self.dbs['__all__']
         try:
-            db.write_points(
+            return db.write_points(
                 points=lines.split('\n')[:-1],
                 database=database,
                 retention_policy=retention_policy,
@@ -185,7 +203,7 @@ def write(self, name, values, **kwargs):
             'fields': values,
             'time': timestamp,
         }
-        self._write(
+        self.store(
             points=[point],
             database=kwargs.get('database') or self.db_name,
             retention_policy=kwargs.get('retention_policy'),
@@ -211,7 +229,7 @@ def batch_write(self, metric_data):
             )
         for database in data_points.keys():
             for rp in data_points[database].keys():
-                self._write(
+                self.store(
                     points=data_points[database][rp],
                     database=database,
                     retention_policy=rp,

diff --git a/openwisp_monitoring/monitoring/migrations/influxdb/influxdb_alter_structure_0006.py b/openwisp_monitoring/monitoring/migrations/influxdb/influxdb_alter_structure_0006.py
@@ -114,10 +114,10 @@ def migrate_influxdb_data(
             start = offset
             end = offset + min(write_data_count, SELECT_QUERY_LIMIT)
             response = retry_until_success(
-                timeseries_db.db.write_points,
+                timeseries_db.store,
                 write_data,
-                tags=metric.tags,
-                batch_size=WRITE_BATCH_SIZE,
+                timeseries_db.db_name,
+                retention_policy=None,
             )
             if response is True:
                 logger.info(