idds | update query for workflow progress page to support throttled requests

core/iDDS/constants.py

@@ -6,7 +6,7 @@
 # http://www.apache.org/licenses/LICENSE-2.0OA
 #
 # Authors:
-# - Wen Guan, <wen.guan@cern.ch>, 2019 - 2022
+# - Wen Guan, <wen.guan@cern.ch>, 2019 - 2023
 
 """
 Constants.
@@ -32,6 +32,8 @@ class Sections:
     Consumer = 'consumer'
     EventBus = 'eventbus'
     Cache = 'cache'
+    Archiver = 'archiver'
+    Coordinator = 'coordinator'
 
 
 class HTTP_STATUS_CODE:
@@ -144,6 +146,7 @@ class RequestStatus(IDDSEnum):
     Terminating = 20
     Building = 21
     Built = 22
+    Throttling = 23
 
 
 class RequestLocking(IDDSEnum):
@@ -288,13 +291,21 @@ class ContentStatus(IDDSEnum):
     Mapped = 7
     FakeAvailable = 8
     Missing = 9
+    Cancelled = 10
+    Activated = 11
 
 
 class ContentLocking(IDDSEnum):
     Idle = 0
     Locking = 1
 
 
+class ContentFetchStatus(IDDSEnum):
+    New = 0
+    Fetching = 1
+    Fetched = 2
+
+
 class GranularityType(IDDSEnum):
     File = 0
     Event = 1
@@ -338,6 +349,12 @@ class ProcessingLocking(IDDSEnum):
     Locking = 1
 
 
+class HealthStatus(IDDSEnum):
+    Default = 0
+    InActive = 1
+    Active = 2
+
+
 class MessageType(IDDSEnum):
     StageInFile = 0
     StageInCollection = 1
@@ -414,6 +431,7 @@ class MessageStatus(IDDSEnum):
     Fetched = 1
     Delivered = 2
     Failed = 3
+    ConfirmDelivered = 4
 
 
 class MessageLocking(IDDSEnum):
@@ -469,6 +487,17 @@ class CommandLocation(IDDSEnum):
     Other = 6
 
 
+class ThrottlerStatus(IDDSEnum):
+    InActive = 0
+    Active = 1
+
+
+class ReturnCode(IDDSEnum):
+    Ok = 0
+    Failed = 255
+    Locked = 1
+
+
 def get_work_status_from_transform_processing_status(status):
     if status in [ProcessingStatus.New, TransformStatus.New]:
         return WorkStatus.New
@@ -487,4 +516,4 @@ def get_work_status_from_transform_processing_status(status):
     elif status in [ProcessingStatus.Terminating, TransformStatus.Terminating]:
         return WorkStatus.Terminating
     else:
-        return WorkStatus.Transforming
+        return WorkStatus.Transforming

core/iDDS/rawsqlquery.py

@@ -98,7 +98,7 @@ def getWorkFlowProgressItemized(request_params, **kwargs):
     left join {settings.DB_SCHEMA_IDDS}.collections c on r.request_id=c.request_id
     left join {settings.DB_SCHEMA_IDDS}.transforms t on t.transform_id=c.transform_id 
     left join {settings.DB_SCHEMA_IDDS}.processings p on p.transform_id=t.transform_id
-    where c.relation_type=0 and {condition} order by r.request_id desc
+    where (c.relation_type=0 or c.relation_type is null) and {condition} order by r.request_id desc
     """
     cur = connections[connection_name].cursor()
     _logger.info('!!! Using connection named: {}, vendor: {}, host: {}, port: {}, user: {} \n Query: {}'.format(

core/iDDS/workflowprogress.py

@@ -32,28 +32,54 @@ def get_workflow_progress_data(request_params, **kwargs):
     workflows_semi_grouped = []
     if not workflows_items.empty:
         workflows_items.USERNAME.fillna(value='', inplace=True)
-        workflows_pd = workflows_items.astype({"WORKLOAD_ID":str}).astype({"R_CREATED_AT":str}).groupby(['REQUEST_ID', 'R_STATUS', 'P_STATUS', 'R_NAME', 'USERNAME']).agg(
+        # workflows_items.PROCESSING_FILES.fillna(value=0, inplace=True)
+        workflows_pd = workflows_items.astype({"WORKLOAD_ID":str}).astype({"R_CREATED_AT":str}).groupby(
+            ['REQUEST_ID', 'R_STATUS', 'P_STATUS', 'R_NAME', 'USERNAME'],
+            dropna=False
+        ).agg(
             PROCESSING_FILES_SUM=pd.NamedAgg(column="PROCESSING_FILES", aggfunc="sum"),
             PROCESSED_FILES_SUM=pd.NamedAgg(column="PROCESSED_FILES", aggfunc="sum"),
             TOTAL_FILES=pd.NamedAgg(column="TOTAL_FILES", aggfunc="sum"),
             P_STATUS_COUNT=pd.NamedAgg(column="P_STATUS", aggfunc="count"),
             R_CREATED_AT=pd.NamedAgg(column="R_CREATED_AT", aggfunc="first"),
             workload_ids=('WORKLOAD_ID', lambda x: '|'.join(x)),
         ).reset_index()
-        workflows_pd = workflows_pd.astype({"R_STATUS":int, 'P_STATUS':int, "PROCESSING_FILES_SUM": int,
-                                            "PROCESSED_FILES_SUM": int, "TOTAL_FILES": int, "P_STATUS_COUNT": int})
+        # fill NAN with 0 for N files
+        workflows_pd.TOTAL_FILES.fillna(value=0, inplace=True)
+        workflows_pd.PROCESSING_FILES_SUM.fillna(value=0, inplace=True)
+        workflows_pd.PROCESSED_FILES_SUM.fillna(value=0, inplace=True)
+        workflows_pd.P_STATUS.fillna(value=0, inplace=True)
+        workflows_pd = workflows_pd.astype({
+            "R_STATUS":int,
+            "P_STATUS":int,
+            "PROCESSING_FILES_SUM": int,
+            "PROCESSED_FILES_SUM": int,
+            "TOTAL_FILES": int,
+            "P_STATUS_COUNT": int
+        })
         workflows_semi_grouped = workflows_pd.values.tolist()
     workflows = {}
     for workflow_group in workflows_semi_grouped:
         workflow = workflows.setdefault(workflow_group[0], {
-            "REQUEST_ID":workflow_group[0], "R_STATUS": subtitleValue.substitleValue("requests", "status")[workflow_group[1]], "CREATED_AT":workflow_group[8],"TOTAL_TASKS":0,
-            "TASKS_STATUSES":{}, "TASKS_LINKS":{}, "REMAINING_FILES":0,"PROCESSED_FILES":0,"PROCESSING_FILES":0,
-            "TOTAL_FILES":0, "TASKS_LINKS_ALL":''})
+            "REQUEST_ID":workflow_group[0],
+            "R_STATUS": subtitleValue.substitleValue("requests", "status")[workflow_group[1]] if \
+                workflow_group[1] in subtitleValue.substitleValue("requests", "status") else \
+                'N/A',
+            "CREATED_AT":workflow_group[8],"TOTAL_TASKS":0,
+            "TASKS_STATUSES":{},
+            "TASKS_LINKS":{},
+            "REMAINING_FILES":0,
+            "PROCESSED_FILES":0,
+            "PROCESSING_FILES":0,
+            "TOTAL_FILES":0,
+            "TASKS_LINKS_ALL":''})
         workflow['TOTAL_TASKS'] += workflow_group[8]
         workflow['R_NAME'] = workflow_group[3]
         workflow['USERNAME'] = workflow_group[4]
         workflow['CREATED_AT'] = workflow_group[9]
-        processing_status_name = subtitleValue.substitleValue("processings", "status")[workflow_group[2]]
+        processing_status_name = subtitleValue.substitleValue("processings", "status")[workflow_group[2]] if \
+            workflow_group[2] in subtitleValue.substitleValue("processings", "status") else \
+            'N/A'
         workflow["TASKS_STATUSES"][processing_status_name] = workflow_group[8]
         workflow["TASKS_LINKS_ALL"] += ('|'+ workflow_group[10].replace('.0','')) if \
             len(workflow["TASKS_LINKS_ALL"]) > 0 else \

core/static/css/state-colors.css

@@ -28,6 +28,7 @@
 .brokeroff { color:#deb900 }
 .holding { color:#FF9933 }
 .throttled { color:#FF9933 }
+.throttling { color:#FF9933 }
 .exhausted { color:#FF9933 }
 .finishing { color:#FF9933 }
 .cancelled { color:#e67300 }
@@ -160,6 +161,7 @@
 
 .holding_fill { color:black; background-color:#FFE38E }
 .throttled_fill { color:black; background-color:#FFE38E; }
+.throttling_fill { color:black; background-color:#FFE38E; }
 .exhausted_fill { color:black; background-color:#FFE38E; }
 
 .pending_fill { color:black;  background-color:#FFF5D6; }