update status notes, error handling and logs

tr_sys/tr_ars/api.py

@@ -55,7 +55,7 @@ def submit(req):
         if 'message' not in data:
             return HttpResponse('Not a valid Translator query json', status=400)
         # create a head message
-        message = Message.create(code=200, status='Done', data=data,
+        message = Message.create(code=200, status='Running', data=data,
                           actor=get_default_actor())
         if 'name' in data:
             message.name = data['name']

tr_sys/tr_ars/default_ars_app/api.py

@@ -2,6 +2,7 @@
 import logging
 import requests
 import sys
+import traceback
 
 from django.http import HttpResponse
 from django.urls import reverse
@@ -43,12 +44,23 @@ def callquery(url, req):
                 mesg = 'Not a valid tr_ars.message'
 
             if data != None:
-                r = query(url, data)
-                resp = HttpResponse(r.text,
-                                     content_type='application/json',
-                                     status=r.status_code)
-                resp['tr_ars.message.status'] = 'R'
-                return resp
+                try:
+                    r = query(url, data)
+                    resp = HttpResponse(r.text,
+                                         content_type='application/json',
+                                         status=r.status_code)
+                    for key in r.headers:
+                        resp['tr_ars.'+key] = r.headers[key]
+                    resp['tr_ars.reason'] = r.reason
+                    resp['tr_ars.url'] = r.url
+                    return resp
+                except:
+                    exc_type, exc_value, exc_traceback = sys.exc_info()
+                    resp = HttpResponse(exc_value,
+                                     status = 503)
+                    resp['tr_ars.url'] = url
+                    return resp
+
     except:
         mesg = 'Unexpected error: %s' % sys.exc_info()
         logger.debug(mesg)

tr_sys/tr_ars/pubsub.py

@@ -6,8 +6,9 @@
 
 logger = logging.getLogger(__name__)
 
+@DeprecationWarning
 def send_message(actor, mesg, timeout=60):
-    url = settings.DEFAULT_HOST + actor.url() # TODO get url base at server startup; no request to use build_absolute_uri()
+    url = settings.DEFAULT_HOST + actor.url()
     logger.debug('sending message %s to %s...' % (mesg.id, url))
     data = mesg.to_dict()
     data['fields']['actor'] = {
@@ -60,7 +61,7 @@ def run(self):
             if settings.USE_CELERY:
                 celery_send_message.delay(actor.to_dict(), mesg.to_dict())
             else:
-                send_message(actor, mesg)
+                celery_send_message(actor.to_dict(), mesg.to_dict())
             queue.task_done()
         logger.debug('%s: BackgroundWorker stopped!' % __name__)
 

tr_sys/tr_ars/signals.py

@@ -12,17 +12,38 @@ def actor_post_save(sender, instance, **kwargs):
     logger.debug('+++ new actor created: %s...%s' % (actor, actor.id))
     # now iterate through each message and pass it to the new actor
     # TODO add flag to turn on/off this behavior
-    send_messages([actor], Message.objects.filter(code=200)
-                  .order_by('timestamp'))
+    # TODO Currently failing because triggered before app is initialized to receive request
+    #send_messages([actor], Message.objects.filter(code=200)
+    #              .order_by('timestamp'))
 
 
 @receiver(post_save, sender=Message)
 def message_post_save(sender, instance, **kwargs):
     message = instance
-    logger.debug('+++ new message created: %s' % (message))
-    # now broadcast the message to all actors only if it has code=200
-    if message.code == 200:
-        send_messages(Actor.objects.all(), [message])
+    logger.debug('+++ message updated: %s' % (message))
+    # now broadcast the message to all actors only if it has code=200 and is a parent node
+    if message.code == 200 and message.ref == None:
+        if len(Message.objects.filter(ref__pk=message.pk)) == 0: # make sure we haven't already done this broadcast
+            send_messages(Actor.objects.all(), [message])
+    # check if parent status should be updated to 'Done'
+    if message.ref and message.status in ['D', 'S', 'E', 'U']:
+        pmessage = message.ref
+        if pmessage.status != 'D':
+            children = Message.objects.filter(ref__pk=pmessage.pk)
+            logger.debug('%s: %d children' % (pmessage.pk, len(children)))
+            finished = True
+            allError = True
+            for child in children:
+                if child.status not in ['D', 'S', 'E', 'U']:
+                    finished = False
+                if child.status not in ['S','E', 'U']:
+                    allError = False
+            if allError or finished:
+                if allError:
+                    pmessage.status = 'E'
+                elif finished:
+                    pmessage.status = 'D'
+                pmessage.save()
 
 @receiver(pre_save, sender=Message)
 def message_pre_save(sender, instance, **kwargs):

tr_sys/tr_ars/status_report.py

@@ -2,6 +2,7 @@
 from .models import Agent, Message, Channel, Actor
 import json, logging, statistics
 import requests
+import datetime
 import Levenshtein
 from django.conf import settings
 
@@ -89,26 +90,23 @@ def status_ars(req, smartresponse, smartapis):
                     if elem[0] == actor['status']:
                         actor['status'] = elem[1]
             data = message.data
-            if 'results' in data:
+            if data is not None and 'results' in data:
                 actor_results = len(data['results'])
             else:
                 actor_results = 0
         if 'status' not in actor:
             actor['status'] = Message.STATUS[-1][1]
         actor_times = []
-        for mesg in Message.objects.filter(actor=a.pk).order_by('-timestamp')[:10]:
+        for mesg in Message.objects.filter(actor=a.pk).order_by('-timestamp')[:3]:
             message = Message.objects.get(pk=mesg.pk)
             parent = Message.objects.get(pk=message.ref.pk)
             actor_times.append((message.timestamp - parent.timestamp).total_seconds())
         actor['results'] = actor_results
         actor['timings'] = 'Unknown'
         if len(actor_times)>0:
-            actor['timings'] = statistics.mean(actor_times)
+            actor['timings'] = str(statistics.mean(actor_times))[:6]
         response['actors'][a.agent.name + '-' + a.path] = actor
 
-    if 'latest' in response['messages']:
-        response['messages']['latest'] = str(latest.timestamp)
-
     # match SmartAPI entries to actors
     matched = []
     for actor in response['actors'].keys():
@@ -142,28 +140,47 @@ def status_ars(req, smartresponse, smartapis):
         else:
             response['actors'][actor]['smartapi'] = "Unknown"
 
+    queue_status = 0 # message queue might be down
+    if 'latest' in response['messages']:
+        tz_info = latest.timestamp.tzinfo
+        diff = datetime.datetime.now(tz_info)-latest.timestamp
+        if diff.total_seconds() < 3:
+            queue_status = 1 # latest is too recent to test whether message queue is down
+        response['messages']['latest'] = str(latest.timestamp)
+
     for key in response['actors'].keys():
         actor = response['actors'][key]
+        #('D', 'Done'),
+        #('S', 'Stopped'),
+        #('R', 'Running'),
+        #('E', 'Error'),
+        #('W', 'Waiting'),
+        #('U', 'Unknown')
+        if actor['status'] not in ['Unknown', 'Error']:
+            queue_status = 2 # message queue is up (as of the last message posted to the queue)
         if actor['status'] in ['Running', 'Done']:
-            if 'smartapireasonercompliant' in actor:
-                if actor['smartapireasonercompliant'] == True:
-                    actor['statusicon'] = 'status2'
-                    actor['statusiconcomment'] = 'up'
-                else:
-                    actor['statusicon'] = 'status8'
-                    actor['statusiconcomment'] = 'SmartAPI incomplete'
-            else:
-                actor['statusicon'] = 'status1'
-                actor['statusiconcomment'] = 'SmartAPI missing'
-        elif 'smartapireasonercompliant' in actor:
+            actor['statusicon'] = 'status2'
+            actor['statusiconcomment'] = 'up'
+        elif actor['status'] in ['Waiting']:
+            actor['statusicon'] = 'status1'
+            actor['statusiconcomment'] = 'Service waiting'
+        elif actor['status'] in ['Error']:
+            actor['statusicon'] = 'status8'
+            actor['statusiconcomment'] = 'Service error'
+        elif actor['status'] in ['Stopped']:
             actor['statusicon'] = 'status9'
-            actor['statusiconcomment'] = 'Service outage'
+            actor['statusiconcomment'] = 'Service stopped'
         else:
             actor['statusicon'] = 'status0'
-            actor['statusiconcomment'] = 'Offline; SmartAPI missing'
+            actor['statusiconcomment'] = 'Service outage'
 
     page = dict()
     page['ARS'] = response
+    page['ARS-Queue-Status'] = queue_status
+    if queue_status == 0:
+        for key in response['actors'].keys():
+            actor = response['actors'][key]
+            actor['statusiconcomment'] = ''
 
     arsreasonsers = dict()
     reasoners = dict()

tr_sys/tr_ars/tasks.py

@@ -5,40 +5,85 @@
 from tr_ars.models import Message, Actor
 from celery.utils.log import get_task_logger
 from django.conf import settings
+import html
 
 logger = get_task_logger(__name__)
 
 @shared_task
 def send_message(actor_dict, mesg_dict, timeout=60):
-    #logger.error(actor_dict)
     logger.info(mesg_dict)
     url = settings.DEFAULT_HOST + actor_dict['fields']['url']
     logger.debug('sending message %s to %s...' % (mesg_dict['pk'], url))
     data = mesg_dict
     data['fields']['actor'] = {
         'id': actor_dict['pk'],
-        'channel': actor_dict['fields']['channel'], #['name'],
-        'agent': actor_dict['fields']['agent'], #['name'],
+        'channel': actor_dict['fields']['channel'],
+        'agent': actor_dict['fields']['agent'],
         'uri': url
     }
     try:
         r = requests.post(url, json=data, timeout=timeout)
         logger.debug('%d: receive message from actor %s...\n%s.\n'
                      % (r.status_code, url, str(r.text)[:500]))
+        url = r.url
+        if 'tr_ars.url' in r.headers:
+            url = r.headers['tr_ars.url']
+        # status defined in https://github.com/NCATSTranslator/ReasonerAPI/blob/master/TranslatorReasonerAPI.yaml
+        # paths: /query: post: responses:
+        # 200 = OK. There may or may not be results. Note that some of the provided
+        #             identifiers may not have been recognized.
+        # 202 = Accepted. Poll /aresponse for results.
+        # 400 = Bad request. The request is invalid according to this OpenAPI
+        #             schema OR a specific identifier is believed to be invalid somehow
+        #             (not just unrecognized).
+        # 500 = Internal server error.
+        # 501 = Not implemented.
+        # Message.STATUS
+        # ('D', 'Done'),
+        # ('S', 'Stopped'),
+        # ('R', 'Running'),
+        # ('E', 'Error'),
+        # ('W', 'Waiting'),
+        # ('U', 'Unknown')
         if r.status_code == 200:
             # now create a new message here
-            status = 'U'
+            status = 'D'
             if 'tr_ars.message.status' in r.headers:
                 status = r.headers['tr_ars.message.status']
-            data = dict()
+            rdata = dict()
             try:
-                data = r.json()
+                rdata = r.json()
             except json.decoder.JSONDecodeError:
                 status = 'E'
             mesg = Message.create(code=r.status_code, status=status,
-                                  data=data, actor=Actor.objects.get(pk=actor_dict['pk']),
-                                  name=mesg_dict['fields']['name'], ref=Message.objects.get(pk=mesg_dict['pk']))
+                                  data=rdata, url=url,
+                                  actor=Actor.objects.get(pk=actor_dict['pk']),
+                                  name=mesg_dict['fields']['name'],
+                                  ref=Message.objects.get(pk=mesg_dict['pk']))
+            mesg.save()
+        else:
+            status = 'U'
+            rdata = data['fields']['data']['message']
+            if r.status_code == 202:
+                status = 'W'
+                url = url[:url.rfind('/')] + '/aresponse/' + r.text
+            if r.status_code >= 400:
+                if r.status_code != 503:
+                    status = 'E'
+                rdata['logs'] = []
+                rdata['logs'].append(html.escape(r.text))
+                for key in r.headers:
+                    if key.lower().find('tr_ars') > -1:
+                        rdata['logs'].append(key+": "+r.headers[key])
+            mesg = Message.create(code=r.status_code, status=status,
+                                  url = url,
+                                  data = rdata,
+                                  actor=Actor.objects.get(pk=actor_dict['pk']),
+                                  name=mesg_dict['fields']['name'],
+                                  ref=Message.objects.get(pk=mesg_dict['pk']))
+            logger.debug('+++ message created: %s' % (mesg.pk))
             mesg.save()
+            logger.debug('+++ message saved: %s' % (mesg.pk))
     except:
         logger.exception("Can't send message to actor %s\n%s"
                          % (url,sys.exc_info()))

tr_sys/tr_ars/templates/status.html

@@ -1,6 +1,21 @@
 {% extends 'ncatspage.html' %}
 {% block content %}
 
+    {% if queue < 2 %}
+
+        <app-ncats-header _ngcontent-c0="" _nghost-c1="">
+            <mat-toolbar _ngcontent-c1="" class="mat-toolbar mat-primary mat-toolbar-multiple-rows">
+                <mat-toolbar-row _ngcontent-c1="" class="mat-toolbar-row" fxlayout="row" id="ncats-logo"
+                                 style="flex-direction: row; box-sizing: border-box; display: flex;">
+                    <div></div>
+                    <div><span style="color: #632f6b;" class="mat-card-title subsection">&nbsp;***
+                        {% if queue == 1 %} Interim timing, please refresh {% else %} Message queue currently down {% endif %} *** </span></div>
+                </mat-toolbar-row>
+            </mat-toolbar>
+        </app-ncats-header>
+
+    {% endif %}
+
     <site-table _ngcontent-c0="" _nghost-c4="">
         <mat-table _ngcontent-c4="" class="mat-table" matsort="" role="grid" ng-reflect-data-source="[object Object]">
             <!---->
@@ -93,6 +108,7 @@
                 </mat-cell>
                 <mat-cell _ngcontent-c4="" class="mat-cell cdk-column-day mat-column-day" role="gridcell">
                     {% if actor.smartapi != 'Unknown' %}<a href="{{ actor.smartapi }}">{{ actor.smartapi|slice:"36:45" }} ...</a>
+                        {% if actor.smartapireasonercompliant == False %} <br> not Reasoner API compliant {% endif %}
                         {% elif actor.smartapiguess %}Is it this <a href="{{ actor.smartapiguess }}">one</a>?
                         {% else %}{{ actor.smartapi }}
                     {% endif %}

tr_sys/tr_ars/views.py

@@ -28,7 +28,8 @@ def status(req):
         'Short_title': 'ARS Status',
         'actors': status['ARS']['actors'],
         'reasoners': status['SmartAPI']['Other-Reasoners'],
-        'sources': status['SmartAPI']['Other-Translator-SmartAPIs']
+        'sources': status['SmartAPI']['Other-Translator-SmartAPIs'],
+        'queue': status['ARS-Queue-Status']
     }
     return HttpResponse(template.render(context, req))