diff --git a/openwisp_monitoring/device/base/models.py b/openwisp_monitoring/device/base/models.py index b3e3f5d4c..ec7b39b06 100644 --- a/openwisp_monitoring/device/base/models.py +++ b/openwisp_monitoring/device/base/models.py @@ -25,7 +25,7 @@ from ...db import device_data_query, timeseries_db from ...monitoring.signals import threshold_crossed -from ...monitoring.tasks import timeseries_write +from ...monitoring.tasks import _timeseries_write from ...settings import CACHE_TIMEOUT from .. import settings as app_settings from .. import tasks @@ -275,7 +275,7 @@ def save_data(self, time=None): self._transform_data() time = time or now() options = dict(tags={'pk': self.pk}, timestamp=time, retention_policy=SHORT_RP) - timeseries_write(name=self.__key, values={'data': self.json()}, **options) + _timeseries_write(name=self.__key, values={'data': self.json()}, **options) cache_key = get_device_cache_key(device=self, context='current-data') # cache current data to allow getting it without querying the timeseries DB cache.set( diff --git a/openwisp_monitoring/monitoring/base/models.py b/openwisp_monitoring/monitoring/base/models.py index f26aec6b7..1a3cdc318 100644 --- a/openwisp_monitoring/monitoring/base/models.py +++ b/openwisp_monitoring/monitoring/base/models.py @@ -35,7 +35,7 @@ ) from ..exceptions import InvalidChartConfigException, InvalidMetricConfigException from ..signals import pre_metric_write, threshold_crossed -from ..tasks import delete_timeseries, timeseries_batch_write, timeseries_write +from ..tasks import _timeseries_batch_write, _timeseries_write, delete_timeseries User = get_user_model() logger = logging.getLogger(__name__) @@ -457,7 +457,7 @@ def write( {'value': extra_values[self.alert_field]} ) if write: - timeseries_write(name=self.key, values=values, **options) + _timeseries_write(name=self.key, values=values, **options) return {'name': self.key, 'values': values, **options} @classmethod @@ -469,7 +469,7 @@ def batch_write(cls, raw_data): write_data.append(metric.write(**kwargs, write=False)) except ValueError as error: error_dict[metric.key] = str(error) - timeseries_batch_write(write_data) + _timeseries_batch_write(write_data) if error_dict: raise ValueError(error_dict) diff --git a/openwisp_monitoring/monitoring/tasks.py b/openwisp_monitoring/monitoring/tasks.py index 9952e73ff..392cb6748 100644 --- a/openwisp_monitoring/monitoring/tasks.py +++ b/openwisp_monitoring/monitoring/tasks.py @@ -39,7 +39,7 @@ def _metric_post_write(name, values, metric, check_threshold_kwargs, **kwargs): autoretry_for=(TimeseriesWriteException,), **RETRY_OPTIONS ) -def _timeseries_write( +def timeseries_write( self, name, values, metric=None, check_threshold_kwargs=None, **kwargs ): """ @@ -49,15 +49,15 @@ def _timeseries_write( _metric_post_write(name, values, metric, check_threshold_kwargs, **kwargs) -def timeseries_write(name, values, metric=None, check_threshold_kwargs=None, **kwargs): +def _timeseries_write(name, values, metric=None, check_threshold_kwargs=None, **kwargs): """ If the timeseries database is using UDP to write data, then write data synchronously. """ if timeseries_db.use_udp: - func = _timeseries_write + func = timeseries_write else: - func = _timeseries_write.delay + func = timeseries_write.delay metric = metric.pk if metric else None func( name=name, @@ -74,7 +74,7 @@ def timeseries_write(name, values, metric=None, check_threshold_kwargs=None, **k autoretry_for=(TimeseriesWriteException,), **RETRY_OPTIONS ) -def _timeseries_batch_write(self, data): +def timeseries_batch_write(self, data): """ Similar to timeseries_write function above, but operates on list of metric data (batch operation) @@ -84,17 +84,17 @@ def _timeseries_batch_write(self, data): _metric_post_write(**metric_data) -def timeseries_batch_write(data): +def _timeseries_batch_write(data): """ If the timeseries database is using UDP to write data, then write data synchronously. """ if timeseries_db.use_udp: - _timeseries_batch_write(data=data) + timeseries_batch_write(data=data) else: for item in data: item['metric'] = item['metric'].pk - _timeseries_batch_write.delay(data=data) + timeseries_batch_write.delay(data=data) @shared_task(base=OpenwispCeleryTask)