Added cleanup function, added timeout to ping plugin

collector.py

@@ -53,8 +53,16 @@ class Collector(object):
                                    start_date=start_date,
                                    name=plugin.__class__.__name__)
 
+    def schedule_cleanup(self):
+        start_date = datetime.now() + timedelta(seconds=100)
+        self.scheduler.add_job(self.cleanup, 'interval', 
+                               hours=24,
+                               start_date=start_date,
+                               name='Cleanup')
+
     def run(self):
         logging.basicConfig()
+        logging.getLogger().setLevel(logging.INFO)
         logging.getLogger('apscheduler').setLevel(logging.INFO)
 
         models = self.collect_models()
@@ -63,6 +71,7 @@ class Collector(object):
             database.DB.create_tables(models)
 
         self.schedule_plugins()
+        self.schedule_cleanup()
         logging.info('Started.')
 
         try:
@@ -72,6 +81,15 @@ class Collector(object):
 
         logging.info(f'Stopped.')
 
+    def cleanup(self):
+        for plugin in self.plugins:
+            logging.info(f'Cleaning up {plugin.__class__.__name__}...')
+            try:
+                items = plugin.cleanup_wrapper()
+                logging.info(f'... deleted {items} entries')
+            except BaseException as e:
+                logging.error("Cleanup error:", exc_info=e)
+                pass
 
 if __name__ == "__main__":
     try:

config.py

@@ -29,19 +29,25 @@ DATABASE_URL = 'postgresql://system_metrics_collector:theMetrixWriteer2123@local
 CPU_INTERVAL = DEFAULT_INTERVAL
 # Store statistics per CPU, not only combined
 CPU_PER_CPU = True
+CPU_RETAIN_DAYS = 180
 
 ### Disk
 DISK_USAGE_INTERVAL = DEFAULT_INTERVAL * 10
+DISK_USAGE_RETAIN_DAYS = 365*50
 DISK_IO_INTERVAL = DEFAULT_INTERVAL
+DISK_IO_RETAIN_DAYS = 180
 
 ### Memory
 MEMORY_INTERVAL = DEFAULT_INTERVAL
+MEMORY_RETAIN_DAYS = 365 * 5
 
 ### Network
 NETWORK_INTERVAL = DEFAULT_INTERVAL
+NETWORK_RETAIN_DAYS = 180
 
 ### Temperatures
 TEMPERATURE_INTERVAL = DEFAULT_INTERVAL
+TEMPERATURE_RETAIN_DAYS = 365 * 5
 
 # If true, fahrenheit is used, otherwise celsius
 TEMPERATURE_USE_FAHRENHEIT = False
@@ -59,8 +65,11 @@ PING_HOSTS = [
     'tibich.com'
 ]
 
+PING_RETAIN_DAYS = 365 * 50
+
 ### Speedtest
-SPEEDTEST_INTERVAL = 15 * 60        # every 15 min
+SPEEDTEST_INTERVAL = 10 * 60        # every 10 min
+SPEEDTEST_RETAIN_DAYS = 365 * 50
 
 ### Stocks
 STOCKS_INTERVAL = 12 * 60 * 60      # updates daily
@@ -75,9 +84,12 @@ STOCKS_TICKERS = {
     'ETH-USD' : 'Ethereum USD',
 }
 
+STOCKS_RETAIN_DAYS = 365 * 50
+
 ### ROBOR
 # Romanian Interbank Offer Rate
 ROBOR_INTERVAL = 12 * 60 * 60       # updates daily, every 12 hours should be fine
+ROBOR_RETAIN_DAYS = 365 * 50
 
 ROBOR_FIELDS = [
     'ROBOR 6M'

plugins/plugin.py

@@ -14,6 +14,13 @@ class Plugin(ABC):
     def execute(self) -> None:
         pass
 
+    def cleanup(self) -> None:
+        return 0
+
     def execute_wrapper(self) -> None:
         with database.DB.connection_context():
-            self.execute()
+            self.execute()
+
+    def cleanup_wrapper(self) -> None:
+        with database.DB.connection_context():
+            return self.cleanup()

plugins/system/cpu_plugin.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -53,3 +53,7 @@ class CpuPlugin(Plugin):
             freqs = psutil.cpu_freq(percpu=True)
             for i in range(len(times)):
                 self.store(i, times[i], freqs[i])
+
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.CPU_RETAIN_DAYS)
+        return Cpu.delete().where(Cpu.time < limit).execute()

plugins/system/disk_plugin.py

@@ -1,5 +1,5 @@
 from collections import namedtuple
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -46,6 +46,10 @@ class DiskUsagePlugin(Plugin):
             entry.free = usage.free
             entry.save()
 
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.DISK_USAGE_RETAIN_DAYS)
+        return DiskUsage.delete().where(DiskUsage.time < limit).execute()
+
 
 class DiskIOPlugin(Plugin):
     models = [DiskIO]
@@ -77,3 +81,6 @@ class DiskIOPlugin(Plugin):
         for disk, current in io_reads.items():
             self.store_io(disk, current)
 
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.DISK_IO_RETAIN_DAYS)
+        return DiskIO.delete().where(DiskIO.time < limit).execute()

plugins/system/memory_plugin.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -46,3 +46,7 @@ class MemoryPlugin(Plugin):
         entry.swap_free = swap.free
         entry.swap_used = swap.used
         entry.save()
+
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.MEMORY_RETAIN_DAYS)
+        return Memory.delete().where(Memory.time < limit).execute()

plugins/system/network_plugin.py

@@ -1,5 +1,5 @@
 from collections import namedtuple
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -48,3 +48,7 @@ class NetworkPlugin(Plugin):
         io_reads = psutil.net_io_counters(pernic=True)
         for nic, current in io_reads.items():
             self.store_io(nic, current)
+
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.NETWORK_RETAIN_DAYS)
+        return NetworkIO.delete().where(NetworkIO.time < limit).execute()

plugins/system/ping_plugin.py

@@ -1,7 +1,7 @@
 import subprocess
 import re
 import subprocess
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -29,19 +29,29 @@ class PingPlugin(Plugin):
 
     def do_ping(self, host):
         command = ['ping', '-c', '1', '-W', str(self.__timeout), host]
-        proc = subprocess.run(command, stdout=subprocess.PIPE)
+        proc = subprocess.Popen(command, stdout=subprocess.PIPE)
-        stdout = proc.stdout.decode()
+        ping = None
+
+        try:
+            proc.wait(60)
+            stdout = proc.stdout.read().decode()
+
+            match = re.search(r'time=([\d\.]+) ms', stdout)
+            if match is not None:
+                ping = float(match.group(1))
+
+        except subprocess.TimeoutExpired:
+            proc.kill()
 
         entry = Ping()
         entry.host = host
-        entry.ping = None
+        entry.ping = ping
-
-        match = re.search(r'time=([\d\.]+) ms', stdout)
-        if match is not None:
-            entry.ping = float(match.group(1))
-
         entry.save()
 
     def execute(self):
         for host in config.PING_HOSTS:
             self.do_ping(host)
+
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.PING_RETAIN_DAYS)
+        return Ping.delete().where(Ping.time < limit).execute()

plugins/system/speedtest_plugin.py

@@ -1,7 +1,7 @@
 import subprocess
 import re
 import subprocess
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -53,5 +53,8 @@ class SpeedtestPlugin(Plugin):
         else:
             logging.error(f"SpeedTest nonzero return: {proc.returncode}\n-----\n{stdout}\n{stderr}\n\n")
             
-
         entry.save()
+
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.SPEEDTEST_RETAIN_DAYS)
+        return Speedtest.delete().where(Speedtest.time < limit).execute()

plugins/system/temperatures_plugin.py

@@ -1,5 +1,5 @@
 from collections import namedtuple
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import psutil
 from peewee import *
@@ -35,3 +35,7 @@ class TemperaturesPlugin(Plugin):
                 entry.high = temp.high
                 entry.critical = temp.critical
                 entry.save()
+
+    def cleanup(self):
+        limit = datetime.utcnow() - timedelta(days=config.TEMPERATURE_RETAIN_DAYS)
+        return Temperatures.delete().where(Temperatures.time < limit).execute()