Merge pull request #27 from imubit/fix-da-broker-svc

Fix da broker svc

Author: cloud-rocket

setup.cfg

@@ -41,14 +41,6 @@ install_requires =
     amqp-fabric
     pywin32 >= 1.0;platform_system=='Windows'
 
-# The usage of test_requires is discouraged, see `Dependency Management` docs
-tests_require =
-    pytest
-    pytest-cov
-    pytest-asyncio
-    aiomisc-pytest
-    coverage
-
 # Require a specific Python version, e.g. Python 2.7 or >= 3.4
 # python_requires = >=2.7,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*
 
@@ -66,7 +58,7 @@ testing =
     setuptools
     pytest
     pytest-cov
-    pytest-asyncio
+    pytest-asyncio==0.26.0
     aiomisc-pytest
 
 [options.entry_points]

src/data_agent/api.py

@@ -321,13 +321,13 @@ def write_manipulated_tags(
 
     # ================== JOBS ===================
     @traceapi
-    def list_jobs(self):
+    def list_jobs(self, conn_name=None):
         """List all running DAQ jobs
 
         :return:
         """
 
-        return self._scheduler.list_jobs()
+        return self._scheduler.list_jobs(conn_name=conn_name)
 
     @traceapi
     def create_job(
@@ -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/broker/agent.py

@@ -22,8 +22,8 @@ class BrokerAgent:
     _data_exchanger = None
     _scheduler = None
 
-    async def init(self, loop, is_service=False, enable_persistance=True):
-        self._config = ConfigManager(loop=loop, enable_persistance=enable_persistance)
+    async def init(self, loop, is_service=False, enable_persistence=True):
+        self._config = ConfigManager(loop=loop, enable_persistence=enable_persistence)
 
         service_config = self._config.get("service")
         broker_config = self._config.get("broker")

src/data_agent/connection_manager.py

@@ -50,6 +50,9 @@ class ConnectionManager:
     def __init__(self, config, extra_connectors=None):
         self._config = config
         self._connections_map = {}
+
+        log.info("Enumerating connection plugins....")
+
         self._connector_classes = {
             entry.name: entry.load() for entry in self.list_plugins()
         }
@@ -59,6 +62,8 @@ def __init__(self, config, extra_connectors=None):
                 self._connector_classes[conn] = extra_connectors[conn]
 
         # Recreate connections from config
+        log.info("Connecting to pre-configured target systems....")
+
         connections = self._config.connections
         for conn in connections:
             self._create_connection(
@@ -68,9 +73,10 @@ def __init__(self, config, extra_connectors=None):
             )
             if connections[conn]["enabled"]:
                 try:
+                    log.info(f"Enabling connection to '{conn}'...")
                     self.enable_connection(conn)
                 except Exception as e:
-                    log.error(f"Error enabling connection {conn} - {e}. ")
+                    log.error(f"Error enabling connection '{conn}' - {e}. ")
                     # self.disable_connection(conn)
 
         log.info(

src/data_agent/connectors/fake_connector.py

@@ -45,9 +45,10 @@ def list_connection_fields():
     def target_info(_):
         return {"Name": "absolute-fake", "Endpoints": []}
 
-    def __init__(self, conn_name="fake_client", **kwargs):
+    def __init__(self, conn_name="fake_client", arg_example="val", **kwargs):
         super(FakeConnector, self).__init__(conn_name)
         self._connected = False
+        self._arg_example = arg_example
         self._tags = {
             "Static": {
                 "Float": {
@@ -99,7 +100,10 @@ def disconnect(self):
         self._groups = {}
 
     def connection_info(self):
-        pass
+        return {
+            "OneLiner": f"[{self.TYPE}] 'absolute-fake'",
+            "ArgExample": self._arg_example,
+        }
 
     @active_connection
     def list_tags(

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]

src/data_agent/win32/config_default.yaml

@@ -72,10 +72,10 @@ log: # standard logging dictConfig
 #          level: 'WARNING',
 #          propagate: False
 
-#    amqp_fabric.amq_broker_connector:
-#      handlers: ['console', 'ntevent', 'file', 'err_file']
-#      level: 'INFO'
-#      propagate: True
+    amqp_fabric.amq_broker_connector:
+      handlers: ['console', 'ntevent', 'file', 'err_file']
+      level: 'INFO'
+      propagate: True
 
     asyncio:
       handlers: ['console', 'err_file']

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_api_target_connection.py

@@ -18,7 +18,9 @@ async def test_lifecycle(rpc_client, rpc_server):
 
     # Create
     assert await rpc_client.proxy.list_connections() == []
-    await rpc_client.proxy.create_connection(conn_name=conn_name, conn_type="fake")
+    await rpc_client.proxy.create_connection(
+        conn_name=conn_name, conn_type="fake", arg_example="me"
+    )
     assert await rpc_client.proxy.list_connections() == [
         {
             "name": "test1",
@@ -65,6 +67,9 @@ async def test_lifecycle(rpc_client, rpc_server):
         }
     ]
 
+    info = await rpc_client.proxy.connection_info(conn_name=conn_name)
+    assert info == {"OneLiner": "[fake] 'absolute-fake'", "ArgExample": "me"}
+
     # Properties
     properties = await rpc_client.proxy.read_tag_attributes(
         conn_name=conn_name, tags=["Random.String", "Random.Real8"]

tests/test_connection_manager.py

@@ -24,7 +24,9 @@ def test_connection_lifecycle(connection_manager):
 
     # Create
     assert connection_manager.list_connections() == []
-    connection_manager.create_connection(conn_name=conn_name, conn_type="fake")
+    connection_manager.create_connection(
+        conn_name=conn_name, conn_type="fake", arg_example="me"
+    )
     assert connection_manager.list_connections() == [
         {
             "name": "test1",