db: Move job and time related processing in api.get_metrics function

devel/test.py

@@ -4,6 +4,7 @@ import requests
 import pytest
 import time
 import json
+import datetime
 
 
 def metrics(params):
@@ -22,13 +23,36 @@ def metrics_available(params={}):
     return r.json()
 
 
-def test_request():
-    assert (
-        len(metrics({"devices": "device-1", "start_time": "2020-12-31T22:59:59"})) > 0
+def test_metrics():
+    assert len(metrics({"devices": "local"})) > 0
+
+
+def test_metrics_start():
+    r = metrics({"devices": "device-1", "start_time": "2020-12-31T23:00:00"})
+    assert len(r) > 0
+    assert all("2020-12-31T23:0" in m["timestamp"] for m in r)
+    assert all("device-1" == m["device_id"] for m in r)
+
+
+def test_metrics_end():
+    r = metrics({"devices": "device-1", "end_time": "2020-12-31T23:19:59"})
+    assert len(r) > 0
+    assert all("2020-12-31T23:1" in m["timestamp"] for m in r)
+
+
+def test_metrics_startend():
+    r = metrics(
+        {
+            "devices": "device-1",
+            "start_time": "2020-12-31T23:00:00",
+            "end_time": "2020-12-31T23:00:59",
+        }
     )
+    assert len(r) > 0
+    assert all("2020-12-31T23:00" in m["timestamp"] for m in r)
 
 
-def test_devices():
+def test_metrics_devices():
     assert 2 * len(
         metrics({"devices": "device-1", "start_time": "2020-12-31T22:59:59"})
     ) == pytest.approx(
@@ -40,29 +64,7 @@ def test_devices():
     )
 
 
-def test_starttime():
-    assert (
-        metrics({"devices": "device-1", "start_time": "2020-12-31T22:59:59.999"})[0][
-            "timestamp"
-        ]
-        == "2020-12-31T23:00:00+00:00"
-    )
-
-
-def test_endtime():
-    assert (
-        metrics(
-            {
-                "devices": "device-1",
-                "start_time": "2020-12-31T22:59:59",
-                "end_time": "2020-12-31T23:30:00.0001",
-            }
-        )[-1]["timestamp"]
-        == "2020-12-31T23:30:00+00:00"
-    )
-
-
-def test_metric():
+def test_metrics_metric():
     assert all(
         r["metric_id"] == "metric-1"
         for r in metrics(
@@ -75,29 +77,40 @@ def test_metric():
     )
 
 
-def test_job():
-    assert all(
-        r["device_id"] in ("device-1", "device-2") for r in metrics({"job_id": 1})
-    )
-    assert metrics({"job_id": 1})[0]["timestamp"] == "2020-12-31T22:00:05+00:00"
-    assert metrics({"job_id": 1})[-1]["timestamp"] == "2020-12-31T22:59:55+00:00"
+def test_metrics_job():
+    r = metrics({"job_id": 1})
+    assert all(m["device_id"] in ("device-1", "device-2") for m in r)
+    assert r[0]["timestamp"] == "2020-12-31T22:00:00+00:00"
+    assert r[-1]["timestamp"] == "2020-12-31T23:00:00+00:00"
 
 
-def test_job_devices():
+def test_metrics_job_devices():
     assert all(
         r["device_id"] == "device-1"
         for r in metrics({"job_id": 1, "devices": "device-1"})
     )
 
 
-def test_job_starttime():
+def test_job_metric():
+    assert all(
+        r["metric_id"] == "metric-1"
+        for r in metrics(
+            {
+                "job_id": 1,
+                "metrics": "metric-1",
+            }
+        )
+    )
+
+
+def test_metrics_job_start():
     assert (
         metrics({"job_id": 1, "start_time": "2020-12-31T22:29:59.999"})[0]["timestamp"]
         == "2020-12-31T22:30:00+00:00"
     )
 
 
-def test_job_endtime():
+def test_metrics_job_end():
     assert (
         metrics(
             {
@@ -120,31 +133,11 @@ def test_job_endtime():
     )
 
 
-def test_job_notfinished():
-    assert (
-        metrics({"job_id": 2, "devices": "device-5",})[
-            -1
-        ]["timestamp"]
-        == "2021-01-01T00:00:00+00:00"
-    )
-    assert (
-        metrics({"job_id": 2, "end_time": "2020-12-31T22:40:00.0001",})[
-            -1
-        ]["timestamp"]
-        == "2020-12-31T22:40:00+00:00"
-    )
-
-
-def test_job_metric():
-    assert all(
-        r["metric_id"] == "metric-1"
-        for r in metrics(
-            {
-                "job_id": 1,
-                "metrics": "metric-1",
-            }
-        )
-    )
+def test_metrics_job_notfinished():
+    r = metrics({"job_id": 3})
+    assert any(str(datetime.date.today()) in m["timestamp"] for m in r)
+    r = metrics({"job_id": 3, "end_time": "2020-12-31T23:59:59"})
+    assert not any(str(datetime.date.today()) in m["timestamp"] for m in r)
 
 
 def test_available_metrics():

kwollect/db/kwollect_setup_db.py

@@ -326,40 +326,47 @@ def time_to_sql(time):
     except ValueError:
         return "'%s'::timestamp with time zone" % time
 
+global start_time, end_time
+
 _devices = ",".join([arg for arg in [nodes,devices] if arg])
 
 if not job_id and not _devices:
     plpy.error("Missing 'devices', 'nodes' or 'job_id' argument")
 
+if job_id:
+    nodetime = plpy.execute("SELECT MIN(start_time) AS start_time, COALESCE(MAX(stop_time), NOW()) AS stop_time, STRING_AGG(node, ',') AS nodes FROM nodetime_by_job WHERE job_id = %s GROUP BY job_id" % job_id)[0]
+    if not start_time:
+        start_time = nodetime["start_time"]
+    # TODO Fix for running job
+    if not end_time:
+        end_time = nodetime["stop_time"]
+    if not _devices:
+        _devices = nodetime["nodes"]
+
 suffix = ""
 if summary:
     suffix = "_summary"
 
 cond = []
-if job_id:
-    req = "SELECT timestamp, device_id, metric_id, value, labels FROM metrics_by_job%s WHERE " % suffix
-    cond.append("job_id = %d" % job_id)
-else:
-    req = "SELECT * FROM metrics_by_device%s WHERE " % suffix
+req = "SELECT * FROM metrics_by_device%s WHERE " % suffix
 
 if _devices:
     _devices = [device.split('.')[0] for device in _devices.split(",")]
     cond.append("device_id in ('%s')" % "','".join(_devices))
 
-end_time_sql = time_to_sql(end_time)
-if job_id and not end_time:
-    pass
-else:
-    cond.append("timestamp <= %s" % end_time_sql)
-
-if job_id and not start_time:
-    pass
+if start_time and end_time:
+    start_time_sql = time_to_sql(start_time)
+    end_time_sql = time_to_sql(end_time)
+elif start_time and not end_time:
+    start_time_sql = time_to_sql(start_time)
+    end_time_sql = time_to_sql(plpy.execute("SELECT %s + INTERVAL '5min' AS t" % start_time_sql)[0]["t"])
+elif not start_time and end_time:
+    end_time_sql = time_to_sql(end_time)
+    start_time_sql = time_to_sql(plpy.execute("SELECT %s - INTERVAL '5min' AS t" % end_time_sql)[0]["t"])
 else:
-    if start_time:
-        start_time_sql = time_to_sql(start_time)
-    else:
-        start_time_sql = end_time_sql + " - interval '5 min'"
-    cond.append("timestamp >= %s" % start_time_sql)
+    end_time_sql = time_to_sql(plpy.execute("SELECT NOW() AS t")[0]["t"])
+    start_time_sql = time_to_sql(plpy.execute("SELECT %s - INTERVAL '5min' AS t" % end_time_sql)[0]["t"])
+cond.append("timestamp >= %s AND timestamp <= %s" % (start_time_sql, end_time_sql))
 
 if metrics:
     metrics_id = metrics.split(",")
@@ -406,7 +413,7 @@ BEGIN
 IF params->>'job_id' IS NOT NULL THEN
   SELECT nullif(substring(params->>'job_id' FROM '^\d+$'), '')::int INTO from_job_id;
   IF from_job_id IS NOT NULL THEN
-    SELECT MAX(stop_time) FROM nodetime_by_job WHERE job_id = from_job_id INTO from_job_id_end;
+    SELECT COALESCE(MAX(stop_time), NOW()) FROM nodetime_by_job WHERE job_id = from_job_id INTO from_job_id_end;
     SELECT ARRAY(SELECT DISTINCT node FROM nodetime_by_job WHERE nodetime_by_job.job_id = from_job_id) INTO from_devices;
     RETURN QUERY SELECT * FROM api.available_metrics(at => LEAST(from_job_id_end, at), params => jsonb_build_object('devices', array_to_json(from_devices)));
     RETURN;