Add job_info API

Author: cloud-rocket

src/data_agent/api.py

@@ -366,6 +366,15 @@ def remove_job(self, job_id: str):
         """
         self._scheduler.remove_job(job_id)
 
+    @traceapi
+    def job_info(self, job_id: str):
+        """Return list of tags in the job
+
+        :return:
+        """
+
+        return self._scheduler.job_info(job_id)
+
     @traceapi
     def list_job_tags(self, job_id: str):
         """Return list of tags in the job

src/data_agent/daq_scheduler.py

@@ -1,7 +1,8 @@
+import datetime as dt
 import json
 import logging
 import time
-from datetime import timedelta
+from datetime import timedelta, timezone
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers import interval
@@ -97,8 +98,15 @@ async def _job_func(self, job_id, conn, broker, tags, from_cache, refresh_rate_m
                 f'Data publish (read time={read_time:.2f}s): {", ".join(to_publish[:120])}'
             )
             broker.publish_data(msg, headers={"job_id": job_id})
+            self._job_state[job_id]["last_successful_timestamp"] = str(
+                dt.datetime.now(timezone.utc).timestamp()
+            )
 
         except Exception as e:
+            self._job_state[job_id]["last_exception_timestamp"] = str(
+                dt.datetime.now(timezone.utc).timestamp()
+            )
+            self._job_state[job_id]["last_exception"] = str(e)
             log.exception(f'Exception in job "{job_id}" - {e}')
 
     def list_jobs(self, conn_name=None):
@@ -206,7 +214,12 @@ def _create_scan_job(self, job_id, conn_name, tags, seconds, from_cache):
             args=[job_id, conn, self._broker_conn, tags, from_cache, refresh_rate_ms],
         )
 
-        self._job_state[job_id] = {"iter_counter": 0}
+        self._job_state[job_id] = {
+            "iter_counter": 0,
+            "last_successful_timestamp": None,
+            "last_exception": None,
+            "last_exception_timestamp": None,
+        }
 
         return job
 
@@ -225,6 +238,24 @@ def remove_job(self, job_id, persist=True):
             if persist:
                 self._config.remove(f"{DAQ_CONFIG_KEY}.{j}")
 
+    def job_info(self, job_id):
+        job = self.get_job(job_id)
+        return {
+            "job_id": job_id,
+            "conn_name": job.args[1].name,
+            "tags": job.args[3],
+            "seconds": int(job.args[5] / 1000),
+            "from_cache": job.args[4],
+            "total_iterations": self._job_state[job_id]["iter_counter"],
+            "last_successful_timestamp": self._job_state[job_id][
+                "last_successful_timestamp"
+            ],
+            "last_exception": self._job_state[job_id]["last_exception"],
+            "last_exception_timestamp": self._job_state[job_id][
+                "last_exception_timestamp"
+            ],
+        }
+
     def list_tags(self, job_id):
         job = self.get_job(job_id)
         return job.args[3]

tests/test_api_daq_job.py

@@ -39,13 +39,29 @@ async def test_job_create_modify(rpc_client, rpc_server, data_queue):
     await rpc_client.proxy.create_job(
         job_id=job1_id, conn_name=conn_name, tags=tags1, seconds=1
     )
+
+    assert await rpc_client.proxy.job_info(job_id=job1_id) == {
+        "job_id": "job1",
+        "conn_name": "test1",
+        "tags": ["Random.Real8", "Random.String"],
+        "seconds": 1,
+        "from_cache": True,
+        "total_iterations": 0,
+        "last_successful_timestamp": None,
+        "last_exception": None,
+        "last_exception_timestamp": None,
+    }
+
     await rpc_client.proxy.create_job(
         job_id=job2_id, conn_name=conn_name, tags=tags2, seconds=1
     )
     assert await rpc_client.proxy.list_jobs() == [job1_id, job2_id]
 
     await asyncio.sleep(1.5)
 
+    info = await rpc_client.proxy.job_info(job_id=job1_id)
+    assert info["total_iterations"] == 1
+
     # Receive 2 messages (1 should be from 1st job and another from 2nd)
     for i in range(2):
         incoming_message = await data_queue.get(timeout=5)

tests/test_daq_scheduler.py

@@ -46,6 +46,18 @@ async def test_job_lifecycle(
     )
     assert scheduler.get_job(job1_id).trigger.interval.seconds == 2
 
+    assert scheduler.job_info(job1_id) == {
+        "job_id": "job1",
+        "conn_name": "fake_conn",
+        "tags": ["Random.String"],
+        "seconds": 2,
+        "from_cache": True,
+        "total_iterations": 0,
+        "last_successful_timestamp": None,
+        "last_exception": None,
+        "last_exception_timestamp": None,
+    }
+
     # Recreate with 1 second nd extra tag
     scheduler.create_scan_job(
         job_id=job1_id,
@@ -56,6 +68,18 @@ async def test_job_lifecycle(
     )
     assert scheduler.get_job(job1_id).trigger.interval.seconds == 1
 
+    assert scheduler.job_info(job1_id) == {
+        "job_id": "job1",
+        "conn_name": "fake_conn",
+        "tags": ["Random.Real8", "Random.String"],
+        "seconds": 1,
+        "from_cache": True,
+        "total_iterations": 0,
+        "last_successful_timestamp": None,
+        "last_exception": None,
+        "last_exception_timestamp": None,
+    }
+
     # Job2
     scheduler.create_scan_job(
         job_id=job2_id, conn_name=conn_name, tags=tags2, seconds=1, from_cache=False