.

Author: ivmfnal

data_dispatcher/db.py

@@ -183,18 +183,20 @@ def as_jsonable(self):
         )
 
 class HasLogRecord(object):
-
-    def __init__(self, log_table, parent_id_columns):
-        self.LogTable = log_table
-        self.IDColumns = parent_id_columns
-        self.IDColumnsText = ",".join(parent_id_columns)
+    
+    #
+    # uses class attributes:
+    #
+    #   LogTable        - name of the table to store the log
+    #   LogIDColumns    - list of columns in the log table identifying the parent
+    #
 
     def add_log(self, type, data=None, **kwargs):
         #print("add_log:", type, data, kwargs)
         c = self.DB.cursor()
         data = (data or {}).copy()
         data.update(kwargs)
-        parent_pk_columns = self.IDColumnsText
+        parent_pk_columns = ",".join(self.LogIDColumns)
         parent_pk_values = ",".join([f"'{v}'" for v in self.pk()])
         c.execute(f"""
             begin;
@@ -203,26 +205,55 @@ def add_log(self, type, data=None, **kwargs):
             commit
         """, (type, json.dumps(data)))
 
+    @sclassmethod
+    def add_log_bulk(cls, db, records):
+        """
+            records: list of tuples:
+                (
+                    (
+                        id_column1_value,
+                        id_column2_value, ...
+                    ),
+                    type,
+                    { data }
+                )
+        """
+        csv = []
+        for id_values, type, data in records:
+            row = '\t'.join([str(v) for v in id_values] + [type, json.dumps(data)])
+            csv.append(row)
+        csv = io.StringIO("\n".join(csv))
+
+        table = cls.LogTable
+        columns = cls.LogIDColumns + ["type", "data"]
+        c = db.cursor()
+        try:
+            c.execute("begin")
+            c.copy_from(csv, table, columns=columns)
+            c.execute("commit")
+        except:
+            c.execute("rollback")
+            raise
+
     def get_log(self, type=None, since=None, reversed=False):
-        parent_pk_columns = self.IDColumnsText
-        parent_pk_values = ",".join(["'{v}'" for v in self.pk()])
+        parent_pk_columns = self.LogIDColumns
+        parent_pk_values = self.pk()
         wheres = [f"{c} = '{v}'" for c, v in zip(parent_pk_columns, parent_pk_values)]
         if isinstance(since, (float, int)):
             since = datetime.utcfromtimestamp(since).replace(tzinfo=timezone.utc)
             wheres.append(f"t >= {since}")
         if type is not None:
             wheres.append(f"type = '{type}'")
-        wheres = "" if not wheres else " and " + " and ".join(wheres)
+        wheres = " and ".join(wheres)
         desc = "desc" if reversed else ""
         sql = f"""
-            select type, t, message from {self.LogTable}
-                {wheres}
+            select type, t, data from {self.LogTable}
+                where {wheres}
                 order by t {desc}
         """
         c = self.DB.cursor()
         c.execute(sql)
-        return (DBLogRecord(type, t, message) for type, t, message in cursor_iterator())
-        
+        return (DBLogRecord(type, t, message) for type, t, message in cursor_iterator(c))
 
 class DBProject(DBObject, HasLogRecord):
 
@@ -232,8 +263,10 @@ class DBProject(DBObject, HasLogRecord):
     Table = "projects"
     PK = ["id"]
 
+    LogIDColumns = ["project_id"]
+    LogTable = "project_log"
+    
     def __init__(self, db, id, owner=None, created_timestamp=None, end_timestamp=None, state=None, retry_count=0, attributes={}):
-        HasLogRecord.__init__(self, "project_log", ["project_id"])
         self.DB = db
         self.ID = id
         self.Owner = owner
@@ -481,8 +514,10 @@ class DBFile(DBObject, HasLogRecord):
     PK = ["namespace", "name"]
     Table = "files"
 
+    LogIDColumns = ["namespace", "name"]
+    LogTable = "file_log"
+    
     def __init__(self, db, namespace, name):
-        HasLogRecord.__init__(self, "file_log", ["namespace", "name"])
         self.DB = db
         self.Namespace = namespace
         self.Name = name
@@ -511,7 +546,10 @@ def replicas(self):
     def create_replica(self, rse, path, url, preference=0, available=False):
         DBReplica.create(self.DB, self.Namespace, self.Name, rse, path, url, preference=preference, available=available)
         self.Replicas = None	# force re-load from the DB
-        
+        self.add_log("found", rse=rse, path=path, url=url)
+        if available:
+            self.add_log("available", rse=rse)
+
     def get_replica(self, rse):
         return self.replicas().get(rse)
 
@@ -659,7 +697,7 @@ def create(db, namespace, name, rse, path, url, preference=0, available=False, e
         except:
             c.execute("rollback")
             raise
-            
+        
         return DBReplica.get(db, namespace, name, rse)
 
     def save(self):
@@ -697,6 +735,19 @@ def remove_bulk(db, rse, dids):
             c.execute("rollback")
             raise
 
+        log_records = (
+            (
+                did.split(":", 1),
+                "removed",
+                {
+                    "rse":  rse,
+                }
+            )
+            for did in dids
+        )
+
+        DBFile.add_log_bulk(db, log_records)
+
     @staticmethod
     def create_bulk(db, rse, preference, replicas):
         # replicas: {(namespace, name) -> {"path":.., "url":..}}
@@ -705,31 +756,25 @@ def create_bulk(db, rse, preference, replicas):
         csv = ['%s\t%s\t%s\t%s\t%s\t%s' % (namespace, name, rse, info["path"], info["url"], preference) 
             for (namespace, name), info in replicas.items()]
         #print("DBReplica.create_bulk: csv:", csv)
-        data = io.StringIO("\n".join(csv))
+        csv = io.StringIO("\n".join(csv))
         table = DBReplica.Table
         columns = DBReplica.columns(as_text=True)
+        
         c = db.cursor()
         try:
             t = int(time.time()*1000)
             temp_table = f"file_replicas_temp_{t}"
             c.execute("begin")
             c.execute(f"create temp table {temp_table} (ns text, n text, r text, p text, u text, pr int)")
-            c.copy_from(data, temp_table)
+            c.copy_from(csv, temp_table)
+            csv = None         # to release memory
             c.execute(f"""
                 insert into {table}({columns}) 
                     select t.ns, t.n, t.r, t.p, t.u, t.pr, false from {temp_table} t
                     on conflict (namespace, name, rse)
                         do nothing;
                 """)
 
-            if False:
-                c.execute(f"""
-                    insert into file_log (namespace, name, type, data)
-                        select t.ns, t.n, 'replica added', ('{"rse":"' || t.r || '", "url":"' || t.u || '", "preference":' || t.pr::text || '}')::jsonb
-                        from {temp_table} t
-                        on conflict (namespace, name, t) do nothing
-                """)
-
             c.execute(f"""
                 drop table {temp_table};
                 commit
@@ -739,6 +784,21 @@ def create_bulk(db, rse, preference, replicas):
             c.execute("rollback")
             raise
 
+        log_records = (
+            (
+                (namespace, name),
+                "found",
+                {
+                    "url":  info["url"],
+                    "rse":  rse,
+                    "path": info["path"]
+                }
+            )
+            for (namespace, name), info in replicas.items()
+        )
+
+        DBFile.add_log_bulk(db, log_records)
+
     @staticmethod
     def update_availability_bulk(db, available, rse, dids):
         # dids is list of dids: ["namespace:name", ...]
@@ -761,6 +821,21 @@ def update_availability_bulk(db, available, rse, dids):
             c.execute("rollback")
             raise
 
+        event = "available" if available else "unavailable"
+        log_records = (
+            (
+                (namespace, name),
+                event,
+                {
+                    "rse":  rse
+                }
+            )
+            for (namespace, name) in undids
+        )
+
+        DBFile.add_log_bulk(db, log_records)
+
+
 class DBFileHandle(DBObject, HasLogRecord):
 
     Columns = ["project_id", "namespace", "name", "state", "worker_id", "attempts", "attributes"]
@@ -780,8 +855,10 @@ class DBFileHandle(DBObject, HasLogRecord):
         ]
 
 
+    LogIDColumns = ["project_id", "namespace", "name"]
+    LogTable = "file_handle_log"
+
     def __init__(self, db, project_id, namespace, name, state=None, worker_id=None, attempts=0, attributes={}):
-        HasLogRecord.__init__(self, "file_handle_log", ["project_id", "namespace", "name"])
         self.DB = db
         self.ProjectID = project_id
         self.Namespace = namespace
@@ -1032,13 +1109,13 @@ def is_active(self):
 
     def done(self):
         self.State = "done"
+        self.add_log("done", worker=self.WorkerID)
         self.WorkerID = None
         self.save()
-        self.add_log("done", worker=self.WorkerID)
 
     def failed(self, retry=True):
         self.State = self.ReadyState if retry else "failed"
-        self.add_log("failed", worker=self.WorkerID or None, retry=retry)
+        self.add_log("failed", worker=self.WorkerID or None, final=not retry)
         self.WorkerID = None
         self.save()
 

docs/ui.rst

@@ -116,11 +116,12 @@ Data Dispatcher provides a way to pass some arbitrary metadata about the project
 The metadata is attached to the project and/or project files at the time of the project creation. Project and file metadata can be any JSON dictionary. 
 If the project is created using a MetaCat query, Data Dispatcher can copy some portions of file metadata from MetaCat to avoid unnecessary
 querying MetaCat at the run time.
-When the worker issues ``dd next -j ...`` command, the output includes the project and the metadata as part of the JSON output.
+When the worker asks for the next file to process, the Data Dispatcher responds with the file information, which includes the project and the 
+file metadata.
 
-Note that project file attributes defined at the project cteation time do not get stored in MetaCat. Also, because file
+Note that the project file attributes defined at the project creation time do not get stored in MetaCat. Also, because file
 attributes are associated with project file handles instead of files, if two projects include the same
-file, they can define project file attributes independently without interfering with each other.
+file, they can define file attributes independently without interfering with each other.
 
 There are several ways to specify project level metadata attributes:
 
@@ -218,10 +219,10 @@ Viewing projects
                 -a                                          - show project attributes only
                 -r                                          - show replicas information
                 -j                                          - show as JSON
-                -f [active|ready|available|all|reserved|failed|done]    - list files (namespace:name) only
+                -f [active|initial|available|all|reserved|failed|done]   - list files (namespace:name) only
                                                                all       - all files, including done and failed
                                                                active    - all except done and failed
-                                                               ready     - ready files only
+                                                               initial   - in initial state
                                                                available - available files only
                                                                reserved  - reserved files only
                                                                failed    - failed files only

web_server/base.html

@@ -211,31 +211,41 @@
 
 
     td.failed {
-        background-color: #F44;
+        background-color: #FCC;
     }
 
     td.done {
-        background-color: #88F;
+        background-color: #CFC;
     }
 
     td.initial {
-        background-color: #DDD;
+        background-color: #EEE;
     }
 
     td.reserved {
-        background-color: #FD8;
+        background-color: #FFC;
     }
 
     td.found {
-        background-color: #ACA;        
+        background-color: #CCF;        
     }
 
     td.available {
-        background-color: #8F8;        
+        background-color: #CFF;             // #AFA        
     }
 
     td.unavailable {
-        background-color: #FAA;        
+        background-color: #FFC;        
+    }
+    
+    a, a:visited {
+        color: #228;
+    }
+    
+    a.button, a.button:visited {
+        color: gray;
+        font-weight: bold;
+        padding: 3px 5px;  
     }
 
     </style>
@@ -267,14 +277,14 @@ <h1>{{GLOBAL_SiteTitle}}</h1>
 <div id="main_menu">
 	<table class="placement link_menu" width="100%">
     	<tr>
-    	    <td><a href="{{GLOBAL_AppTopPath}}/P/projects">projects</a></td>
-    	    <td><a href="{{GLOBAL_AppTopPath}}/R/index">RSEs</a></td>
+    	    <td><a class=button href="{{GLOBAL_AppTopPath}}/P/projects">projects</a></td>
+    	    <td><a class=button href="{{GLOBAL_AppTopPath}}/R/index">RSEs</a></td>
             <td style="width:100%"></td>
     		<td>
                         {% if GLOBAL_User %}
-                            <span style="color:#AAA">logged in as {{GLOBAL_User.Name or ""}}&nbsp;<a href="{{GLOBAL_AppTopPath}}/U/user?username={{GLOBAL_User.Username}}">{{GLOBAL_User.Username}}</a>&nbsp;</span><a href="{{GLOBAL_AppTopPath}}/A/logout?redirect={{GLOBAL_AppTopPath}}/index">log out</a>
+                            <span style="color:#AAA">logged in as {{GLOBAL_User.Name or ""}}&nbsp;<a href="{{GLOBAL_AppTopPath}}/U/user?username={{GLOBAL_User.Username}}">{{GLOBAL_User.Username}}</a>&nbsp;</span><a class=button  href="{{GLOBAL_AppTopPath}}/A/logout?redirect={{GLOBAL_AppTopPath}}/index">log out</a>
                         {% else %}
-                            <a href="{{GLOBAL_AppTopPath}}/A/login?redirect={{GLOBAL_AppTopPath}}/index">log in</a>
+                            <a class=button href="{{GLOBAL_AppTopPath}}/A/login?redirect={{GLOBAL_AppTopPath}}/index">log in</a>
                         {% endif %}
     		</td>
     	</tr>

web_server/gui_server.py

@@ -1,9 +1,9 @@
 from webpie import WPApp, WPHandler
-from data_dispatcher.db import DBProject, DBFileHandle, DBRSE
+from data_dispatcher.db import DBProject, DBFileHandle, DBRSE, DBUser
 from data_dispatcher import Version
 from metacat.auth.server import AuthHandler, BaseHandler, BaseApp
 import urllib, os, yaml
-from urllib.parse import quote, unquote
+from urllib.parse import quote, unquote, unquote_plus
 from wsdbtools import ConnectionPool
 
 class UsersHandler(BaseHandler):
@@ -152,7 +152,7 @@ def handle(self, request, relpath, project_id=None, namespace=None, name=None, *
         handle = DBFileHandle.get(db, int(project_id), namespace, name)
         if handle is None:
             self.redirect(f"./project?project_id={project_id}&error=Handle+not+found")
-        return self.render_to_response("handle.html", project_id=project_id, handle=handle)
+        return self.render_to_response("handle.html", project_id=project_id, handle=handle, handle_log = list(handle.get_log(reversed=True)))
 
 
 class RSEHandler(BaseHandler):