"""
File name: algorithms.py
Author: Nico Vermaas - Astron
Date created: 2019-04-04
Description: Business logic for ATDB. These functions are called from the views (views.py).
"""
from django.db.models import Q,Sum
import logging
from .common import timeit
from ..models import Task, LogEntry, Workflow
from django.conf import settings
DATE_FORMAT = "%Y-%m-%d"
TIME_FORMAT = "%Y-%m-%d %H:%M:%SZ"
DJANGO_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
logger = logging.getLogger(__name__)
@timeit
def get_size(status_list):
"""
aggregate the sizes of all task with a status in the list
:param status_list: list of statuses to consider for the aggregation
:return: summed sizes
"""
logger.info("get_size("+str(status_list)+")")
field = 'size_to_process'
query = field + '__sum'
tasks = Task.objects.filter(status__in=status_list)
sum_value = tasks.aggregate(Sum(field))[query]
if sum_value == None:
sum_value = 0.0
return sum_value
def convert_logentries_to_html(log_entries):
results = ""
try:
results += "<th>service</th><th>status</th><th>timestamp</th><th>cpu_cycles</th><th>wall_clock_time</th><th>logfile</th>"
results += "<tbody>"
for log in log_entries:
line = "<tr><td><b>" + log.step_name + '</b></td>'
line +='<td class="' + log.status + '" >' + log.status + "</td>"
line += "<td>" + str(log.timestamp.strftime("%m-%d-%Y, %H:%M:%S")) + "</td>"
line += "<td>" + str(log.cpu_cycles) + "</td>"
line += "<td>" + str(log.wall_clock_time) + "</td>"
if log.url_to_log_file!=None:
link = "<a href=" + '"' + str(log.url_to_log_file) + '" target="_blank">' + "logfile" + "</a>"
else:
link = "-"
line += "<td>" + link + "</td>"
results += line
results += "</tbody>"
except:
results = "<tr><td>no data</td></tr>"
return results
def convert_list_of_dicts_to_html(my_list):
results = ""
try:
for my_dict in my_list:
# iterate through the dict of key/values
for key,value in my_dict.items():
try:
if "://" in value:
link = "<a href=" + '"' + value + '">' + key +"</a>"
value = link
except:
pass
line = "<tr><td><b>" + str(key) + "</b></td><td>" + str(value) + "</td></tr>"
results = results + line
except:
results = "<tr><td>no data</td></tr>"
return results
def convert_config_to_html(querylist):
results = ""
try:
for record in querylist:
# iterate through the dict of key/values
key = record.key
value = record.value
try:
if "://" in value:
link = "<a href=" + '"' + value + '">' + key +"</a>"
value = link
except:
pass
line = "<tr><td><b>" + str(key) + "</b></td><td>" + str(value) + "</td></tr>"
results = results + line
except:
results = "<tr><td>no data</td></tr>"
return results
# aggregate information from the tasks table per workflow per status
def aggregate_resources_tasks():
workflow_results = []
# get all active tasks
active_tasks = Task.objects.filter(status__in=settings.ACTIVE_STATUSSES)
# active_tasks_count = active_tasks.count()
# retrieve all unique workflows
active_workflows = active_tasks.values('workflow').distinct()
# iterate through the filters and accumulate logentries
for w in active_workflows:
workflow_result = {}
# extract the workflow object (cheap)
workflow = Workflow.objects.get(id = w['workflow'])
# get the numbers for this workflow
# all tasks for this workflow for the 'grand total'
tasks_per_workflow = Task.objects.filter(workflow=workflow)
nr_of_tasks_per_workflow = tasks_per_workflow.count()
sum_size_to_process = tasks_per_workflow.aggregate(Sum('size_to_process'))
workflow_result['size_to_process'] = sum_size_to_process['size_to_process__sum']
sum_size_processed = tasks_per_workflow.aggregate(Sum('size_processed'))
workflow_result['size_processed'] = sum_size_processed['size_processed__sum']
# all the active tasks
active_tasks_per_workflow = tasks_per_workflow.filter(status__in=settings.ACTIVE_STATUSSES)
nr_of_active_tasks_per_workflow = active_tasks_per_workflow.count()
# split per status, to see the progress
nr_per_status = {}
for status in settings.ALL_STATUSSES:
record = {}
nr_for_this_status = Task.objects.filter(workflow=workflow, status=status).count()
record[status] = nr_for_this_status
nr_per_status[status] = nr_for_this_status
nr_per_status['active'] = nr_of_active_tasks_per_workflow
nr_per_status['total'] = nr_of_tasks_per_workflow
# store the results in a dict
workflow_result['id'] = workflow.id
workflow_result['name'] = workflow.workflow_uri
workflow_result['nr_of_tasks'] = nr_of_tasks_per_workflow
workflow_result['nr_of_active_tasks'] = nr_of_active_tasks_per_workflow
workflow_result['nr_of_tasks_per_status'] = nr_per_status
workflow_results.append(workflow_result)
return workflow_results
# aggregate information from the logentries table per workflow per status
def aggregate_resources_logs():
records = []
# get all active tasks
active_tasks = Task.objects.filter(status__in=settings.ACTIVE_STATUSSES)
active_tasks_count = active_tasks.count()
# retrieve all unique workflows
active_workflows = active_tasks.values('workflow').distinct()
# iterate through the filters and accumulate logentries
for w in active_workflows:
workflow_result = {}
# extract the workflow object (cheap)
workflow = Workflow.objects.get(id = w['workflow'])
# aggregate logentries per step for all active statusses
for status in settings.ACTIVE_STATUSSES:
record = {}
record['name'] = str(workflow.id) +' - '+ workflow.workflow_uri
record['status'] = status
# aggregate logentries per step for all active statusses (expensive)
logs = LogEntry.objects.filter(status=status)\
.filter(task__status__in=settings.ACTIVE_STATUSSES)\
.filter(task__workflow=workflow)
sum_cpu_cycles = logs.aggregate(Sum('cpu_cycles'))
record['cpu_cycles'] = sum_cpu_cycles['cpu_cycles__sum']
wall_clock_time = logs.aggregate(Sum('wall_clock_time'))
record['wall_clock_time'] = wall_clock_time['wall_clock_time__sum']
records.append(record)
return records
def construct_link(request, status, workflow_id, count):
link = str(count)
try:
query = "?status=" + status + "&workflow__id=" + str(workflow_id)
if settings.DEV == True:
url = request.build_absolute_uri('/atdb/tasks') + query
else:
# Unclear why 'build_absolute_uri' doesn't return 'https' in production.
# Probably because the https is handled fully outside the container by Traefik
# and ATDB is not aware of that.
url = "https://" + request.get_host() + '/atdb/tasks' + query
link = '<a href="' + url + '" target="_blank">' + str(count) + "</a>"
except:
pass
return link
def human_readable(size_in_bytes):
for count in ['Bytes', 'KB', 'MB', 'GB', 'TB']:
if size_in_bytes > -1024.0 and size_in_bytes < 1024.0:
return "%3.1f %s" % (size_in_bytes, count)
size_in_bytes /= 1024.0
return "%3.1f %s" % (size_in_bytes, 'PB')
def construct_tasks_per_workflow_html(request, workflow_results):
# --- Progress of tasks per active workflow ---
workflow_results = aggregate_resources_tasks()
results_tasks = "<p>Progress of tasks per (active) workflow</p>"
header = "<th>Workflow</th>"
for status in settings.ALL_STATUSSES:
header += "<th>" + status + "</th>"
results_tasks += header + '<th class="active">active</th><th>total</th><th>to process</th><th>processed</th>'
for workflow_result in workflow_results:
d = workflow_result['nr_of_tasks_per_status']
#values = "<td><b>" + str(workflow_result['id'])+" - "+workflow_result['name'] + "</b></td>"
values = "<td><b>" + str(workflow_result['id']) + "</b></td>"
for key in d:
percentage = round(int(d[key]) / int(workflow_result['nr_of_tasks']) * 100)
# distinguish active statusses
style = ""
if key in settings.ACTIVE_STATUSSES or key=='active':
style = "active"
# bonus: add a query link
link = construct_link(request, key, workflow_result['id'], d[key])
values += "<td class=" + style + ">" + str(percentage) + "% (" + link + ")</td>"
#values += "<td class="+style+">" + str(percentage) + "% ("+str(d[key])+")</td>"
#values += "<td>" + str(d[key]) + "</td>"
# add sizes
values += "<td>" + str(human_readable(workflow_result['size_to_process'])) + "</td>"
try:
percentage = round(int(workflow_result['size_processed']) / int(workflow_result['size_to_process']) * 100)
except:
percentage = 0
values += "<td>" + str(human_readable(workflow_result['size_processed'])) + " ("+ str(percentage) + "%) </td>"
results_tasks += "<tr>" + values + "</tr>"
results_tasks = "<tbody>" + results_tasks + "</tbody>"
return results_tasks
def construct_logs_per_workflow_html(log_records):
results_logs = ""
for record in log_records:
# distinguish active statusses
style = ""
if record['status'] in settings.ACTIVE_STATUSSES:
style = "active"
line = "<tr><td><b>" + record['name'] + "</b></td>" \
'<td class="' + style + '" >' + record['status'] + \
"</td><td>" + str(record['cpu_cycles']) + \
"</td><td>" + str(record['wall_clock_time']) + "</td><tr>"
results_logs += line
return results_logs
def construct_dashboard_html(request):
# --- Progress of tasks per active workflow ---
workflow_results = aggregate_resources_tasks()
results_tasks = construct_tasks_per_workflow_html(request, workflow_results)
# --- logentries ---
log_records = aggregate_resources_logs()
results_logs = construct_logs_per_workflow_html(log_records)
return results_tasks,results_logs