Process queue

app.py

@@ -2,13 +2,16 @@ import fcntl
 import hmac
 import json
 import logging
+import logging.config
 import os
 
 from flask import (Flask, request, jsonify, abort, render_template)
 
 from ltsdb_json import LTS
 
-logging.basicConfig(level=logging.DEBUG)
+import config
+
+logging.config.dictConfig(config.logging)
 
 app = Flask(__name__)
 

ltsdb_json.py

@@ -9,6 +9,7 @@ import time
 
 class LTS:
     base_dir = "data"
+    queue_dir = "queue"
     limit = 1000
 
     def __init__(self, description=None, id=None):
@@ -20,6 +21,7 @@ class LTS:
             m.update(bytes(serialized_description, encoding="UTF-8"))
             id = m.hexdigest()
         self.filename = self.base_dir + "/" + id
+        self.id = id
         try:
             with open(self.filename, "r") as fh:
                 fcntl.flock(fh, fcntl.LOCK_SH)
@@ -57,6 +59,8 @@ class LTS:
             fcntl.flock(fh, fcntl.LOCK_EX)
             json.dump({"description": self.description, "data": self.data}, fh)
             fh.truncate()
+        with open(self.queue_dir + "/" + self.id, "w") as fh:
+            pass
 
     def rebuild_index(self):
         t0 = time.time()

process_queue

@@ -0,0 +1,71 @@
+#!/usr/bin/python3
+
+import os
+import socket
+import time
+
+from ltsdb_json import LTS
+
+node = socket.gethostbyaddr(socket.gethostname())[0]
+
+class DiskFullPredictor:
+    def match(self, lts):
+        # measure=bytes_used, mountpoint=*
+        if "measure" not in lts.description:
+            return False
+        if lts.description["measure"] != "bytes_used":
+            return False
+        if "mountpoint" not in lts.description:
+            return False
+        return True
+
+    def run(self, lts):
+        # find matching bytes_usable series
+        desc = {**lts.description, "measure": "bytes_usable"}
+        usable_lts = LTS(description=desc)
+        # The two timeseries are always updated together, so the
+        # timestamps should match exactly. But just in case we decide to
+        # change that in the future accept a difference of up to an
+        # hour.
+        now = lts.data[-1][0]
+        if abs(now - usable_lts.data[-1][0]) > 3600:
+            log.warning("Timeseries %s and %s have different end times: %s vs %s",
+                        lts.id, usable_lts.id,
+                        now, usable_lts.data[-1][0])
+            return
+        current_used_bytes = lts.data[-1][1]
+        current_usable_bytes = usable_lts.data[-1][1]
+        tuf = float('inf')
+        for d in reversed(lts.data):
+            if d[1] < current_usable_bytes * 0.1:
+                continue # for sanity
+            if current_used_bytes ** 2 / d[1] > current_usable_bytes:
+                tuf = now - d[0]
+                break
+        desc = {**lts.description,
+             "measure": "time_until_disk_full",
+             "node": node,
+             "unit": "s",
+             "remote_addr": "",
+            }
+        lts = LTS(desc)
+        lts.add(now, tuf)
+        lts.save()
+
+processors = [
+    DiskFullPredictor(),
+]
+
+def process(lts):
+    for processor in processors:
+        if processor.match(lts):
+            processor.run(lts)
+
+while True:
+    for id in os.listdir("queue"):
+        lts = LTS(id=id)
+        os.remove("queue/" + id)
+        process(lts)
+    time.sleep(1)
+
+