LIU-470: Per-app logging from the web UI

- Setup a logging handler for the AppDrop class
- Add REST endpoint in the DropManagers to get collected logs for each AppDROP.
- Add basic clickable link to REST API call in the DIM web UI

Author: myxie

daliuge-common/dlg/clients.py

@@ -118,6 +118,16 @@ def graph_status(self, sessionId):
         )
         return ret
 
+    def get_drop_status(self, sid, did):
+        ret = self._get_json(f"/sessions/{quote(sid)}/graph/drop/{quote(did)}")
+        logger.debug(
+            "Successfully read graph status from session %s on %s:%s",
+            sid,
+            self.host,
+            self.port,
+        )
+        return ret
+
     def graph(self, sessionId):
         """
         Returns a dictionary where the key are the DROP UIDs, and the values are
@@ -220,6 +230,7 @@ def graph_size(self, sessionId):
     getGraphStatus = graph_status
     getGraphSize = graph_size
     getGraph = graph
+    getDropStatus = get_drop_status
 
 
 class NodeManagerClient(BaseDROPManagerClient):

daliuge-engine/dlg/apps/app_base.py

@@ -2,6 +2,7 @@
 from collections import OrderedDict
 from concurrent.futures import Future
 from typing import List, Callable
+import json
 import logging
 import math
 import threading
@@ -73,7 +74,25 @@ def thread_target():
 
 _SYNC_DROP_RUNNER = SyncDropRunner()
 
-
+class InstanceLogHandler(logging.Handler):
+    """Custom handler to store logs in-memory per object instance."""
+    def __init__(self, log_storage):
+        super().__init__()
+        self.log_storage = log_storage
+
+    def emit(self, record):
+        """Store log messages in the instance's log storage."""
+        log_entry = self.format(record)
+        self.log_storage.append(log_entry)
+
+class AppLogFilter(logging.Filter):
+    def __init__(self, uid: str, humanKey: str):
+        self.uid = uid
+        self.humanKey = humanKey
+
+    def filter(self, record):
+        uid = getattr(record, "drop_uid", None)
+        return uid == self.uid or uid == self.humanKey
 # ===============================================================================
 # AppDROP classes follow
 # ===============================================================================
@@ -141,6 +160,18 @@ def initialize(self, **kwargs):
         # by default run drops synchronously
         self._drop_runner: DropRunner = _SYNC_DROP_RUNNER
 
+        self.log_storage = []
+
+        self.logger = logging.getLogger(f"{__class__}.{self.uid}")
+        instance_handler = InstanceLogHandler(self.log_storage)
+        instance_handler.addFilter(AppLogFilter(self.uid, self._humanKey))
+
+        # Attach instance-specific handler
+        logging.root.addHandler(instance_handler)
+
+        # Ensure logs still propagate to the root logger
+        logger.propagate = True
+
     @track_current_drop
     def addInput(self, inputDrop, back=True):
         uid = inputDrop.uid
@@ -308,6 +339,14 @@ def skip(self):
                 execStatus=self.execStatus,
             )
 
+    def getLogs(self):
+        """
+        :return: Return the logs stored in the logging handler
+        """
+
+        return self.log_storage
+
+
 
 class InputFiredAppDROP(AppDROP):
     """

daliuge-engine/dlg/manager/composite_manager.py

@@ -306,7 +306,7 @@ def getPastSessionIds(self) -> list[str]:
             for path in self._past_session_manager.past_sessions(self._sessionIds)
         ]
 
-    def _do_in_host(self, action, sessionId, exceptions, f, collect, port, iterable):
+    def _do_in_host(self, action, sessionId, exceptions, f, collect, port, iterable, **kwargs):
         """
         Replication of commands to underlying drop managers
         If "collect" is given, then individual results are also kept in the given
@@ -324,7 +324,7 @@ def _do_in_host(self, action, sessionId, exceptions, f, collect, port, iterable)
 
         try:
             with self.dmAt(host) as dm:
-                res = f(dm, iterable, sessionId)
+                res = f(dm, iterable, sessionId, **kwargs)
 
             if isinstance(collect, dict):
                 collect.update(res)
@@ -342,7 +342,8 @@ def _do_in_host(self, action, sessionId, exceptions, f, collect, port, iterable)
                 f,
             )
 
-    def replicate(self, sessionId, f, action, collect=None, iterable=None, port=None):
+    def replicate(self, sessionId, f, action, collect=None, iterable=None, port=None,
+                  **kwargs):
         """
         Replicates the given function call on each of the underlying drop managers
         """
@@ -352,7 +353,8 @@ def replicate(self, sessionId, f, action, collect=None, iterable=None, port=None
         logger.debug("Replicating command: %s on hosts: %s", f, iterable)
         self._tp.map(
             functools.partial(
-                self._do_in_host, action, sessionId, thrExs, f, collect, port
+                self._do_in_host, action, sessionId, thrExs, f, collect, port,
+                **kwargs
             ),
             iterable,
         )
@@ -598,6 +600,30 @@ def getGraphStatus(self, sessionId):
         )
         return allStatus
 
+    def getDropStatus(self, sessionId, dropId):
+        allstatus = {}
+        self.replicate(
+            sessionId,
+            # {"session": sessionId, "drop": dropId},
+            self._getDropStatus,
+            "getting graph status",
+            collect=allstatus,
+            dropId=dropId
+        )
+        return allstatus
+
+    def _getDropStatus(self, dm, host, sessionId, dropId ):
+        """
+        See session.getDropLogs()
+
+        :param dm:
+        :param host:
+        :param sessionId:
+        :param dropId:
+        :return: JSON of status logs and DROP information
+        """
+        return dm.getDropStatus(sessionId, dropId)
+
     def _getGraph(self, dm, host, sessionId):
         return dm.getGraph(sessionId)
 

daliuge-engine/dlg/manager/node_manager.py

@@ -364,6 +364,9 @@ def getGraphStatus(self, sessionId):
         self._check_session_id(sessionId)
         return self._sessions[sessionId].getGraphStatus()
 
+    def getDropStatus(self, sessionId, dropId):
+        return self._sessions[sessionId].getDropLogs(dropId)
+
     def getGraph(self, sessionId):
         self._check_session_id(sessionId)
         #  TODO: Ensure returns reproducibility data.

daliuge-engine/dlg/manager/rest.py

@@ -205,6 +205,7 @@ def __init__(self, dm, maxreqsize=10):
             method="OPTIONS",
             callback=self.acceptPreflight2,
         )
+        app.get("/api/sessions/<sessionId>/graph/drop/<dropId>", callback=self.getDropStatus)
 
         # The non-REST mappings that serve HTML-related content
         app.route("/static/<filepath:path>", callback=self.server_static)
@@ -334,6 +335,10 @@ def getGraphSize(self, sessionId):
     def getGraphStatus(self, sessionId):
         return self.dm.getGraphStatus(sessionId)
 
+    @daliuge_aware
+    def getDropStatus(self, sessionId, dropId):
+        return self.dm.getDropStatus(sessionId, dropId)
+
     @daliuge_aware
     def addGraphSpec(self, sessionId):
         # WARNING: TODO: Somehow, the content_type can be overwritten to 'text/plain'

daliuge-engine/dlg/manager/session.py

@@ -642,6 +642,18 @@ def getGraphStatus(self):
 
         return statusDict
 
+    def getDropLogs(self, drop_oid: str):
+        """
+        Retrieve the logs stored in the given DROP
+        :param drop_oid: drop_oid
+
+        :return:
+        """
+        return {"session": self.sessionId,
+                "status": self.status,
+                "logs": self._drops[drop_oid].getLogs() }
+
+
     @track_current_session
     def cancel(self):
         status = self.status

daliuge-engine/dlg/manager/web/session.html

@@ -95,7 +95,7 @@
         d3.select('#pg-graph').remove();
 
 	const width = $('#pg-progress-bar').innerWidth();
-	var graph_update_handler = function(oids, dropSpecs) {};
+	var graph_update_handler = function(oids, dropSpecs, url) {};
 
 	var status_update_handler = function(statuses){
 		// This is the order in which blocks are drawn in the progress bar,

daliuge-engine/dlg/manager/web/static/js/dm.js

@@ -457,7 +457,7 @@ function promptNewSession(serverUrl, tbodyEl, refreshBtn) {
 	});
 }
 
-function drawGraphForDrops(g, drawGraph, oids, doSpecs) {
+function drawGraphForDrops(g, drawGraph, oids, doSpecs, url) {
 
 	// Keep track of modifications to see if we need to re-draw
 	var modified = false;
@@ -468,7 +468,7 @@ function drawGraphForDrops(g, drawGraph, oids, doSpecs) {
 	for (var idx in oids) {
 		if (oids[idx] != 'reprodata') {
 			var doSpec = doSpecs[oids[idx]];
-			modified |= _addNode(g, doSpec);
+			modified |= _addNode(g, doSpec, url);
 		}
 	}
 
@@ -572,7 +572,7 @@ function startStatusQuery(serverUrl, sessionId, selectedNode, graph_update_handl
 			if (oids.length > 0) {
 				// Get sorted oids
 				oids.sort();
-				graph_update_handler(oids, doSpecs);
+				graph_update_handler(oids, doSpecs, url);
 			}
 
 			// During PRISITINE and BUILDING we need to update the graph structure
@@ -603,7 +603,7 @@ function startStatusQuery(serverUrl, sessionId, selectedNode, graph_update_handl
 	var updateGraphTimer = d3.timer(updateGraph);
 }
 
-function _addNode(g, doSpec) {
+function _addNode(g, doSpec, url) {
 
 	if (g.hasNode(g)) {
 		return false;
@@ -626,13 +626,16 @@ function _addNode(g, doSpec) {
 	else if (doSpec.type == 'plain') {
 		notes += 'storage: ' + doSpec.storage;
 	}
-
+    url = url + "/graph/drop/" +  doSpec.oid;
+//    let link = "<a href=" + url +  target='_blank'>Click Me</a>";
+    let link = "<a href=" + url +  " target='_blank'>Click Me</a>";
 	var oid = doSpec.oid;
 	var html = '<div class="drop-label ' + typeShape + '" id="id_' + oid + '">';
 	html += '<span class="notes">' + notes + '</span>';
     oid_date = doSpec.oid.split("_")[0];
 	human_readable_id = oid_date + "_" + doSpec.humanReadableKey.toString()
 	html += '<span style="font-size: 13px;">' + human_readable_id + '</span>';
+	html += '<span style="font-size: 13px;">' + link + '</span>';
 	html += "</div>";
 	g.setNode(oid, {
 		labelType: "html",