Repository: rucio/rucio
Issue: Hermes2: Add a protection against empty transferred_at

Motivation
----------
When someone cancels a transfer, `transferred_at` is not set which crashes hermes2 if it works on influx.




Changed files:
lib/rucio/daemons/hermes/hermes2.py

diff --git a/lib/rucio/daemons/hermes/hermes2.py b/lib/rucio/daemons/hermes/hermes2.py
index 363986e8e..0da494360 100644
--- a/lib/rucio/daemons/hermes/hermes2.py
+++ b/lib/rucio/daemons/hermes/hermes2.py
@@ -39,7 +39,6 @@ import socket
 import sys
 import threading
 import time
-import traceback
 from email.mime.text import MIMEText
 
 import requests
@@ -50,7 +49,7 @@ from six import PY2
 import rucio.db.sqla.util
 from rucio.common.config import config_get, config_get_int, config_get_bool
 from rucio.common.exception import ConfigNotFound, DatabaseException
-from rucio.common.logging import setup_logging
+from rucio.common.logging import setup_logging, formatted_logger
 from rucio.core import heartbeat
 from rucio.core.config import get
 from rucio.core.message import retrieve_messages, delete_messages, update_messages_services
@@ -86,8 +85,14 @@ class HermesListener(stomp.ConnectionListener):
         logging.error('[broker] [%s]: %s', self.__broker, frame.body)
 
 
-def setup_activemq(prepend_str):
-    logging.info('%s [broker] Resolving brokers', prepend_str)
+def setup_activemq(logger):
+    """
+    Deliver messages to ActiveMQ
+
+    :param logger:             The logger object.
+    """
+
+    logger(logging.INFO, '[broker] Resolving brokers')
 
     brokers_alias = []
     brokers_resolved = []
@@ -96,31 +101,31 @@ def setup_activemq(prepend_str):
     except:
         raise Exception('Could not load brokers from configuration')
 
-    logging.info('%s [broker] Resolving broker dns alias: %s', prepend_str, brokers_alias)
+    logger(logging.INFO, '[broker] Resolving broker dns alias: %s', brokers_alias)
     brokers_resolved = []
     for broker in brokers_alias:
         try:
             addrinfos = socket.getaddrinfo(broker, 0, socket.AF_INET, 0, socket.IPPROTO_TCP)
             brokers_resolved.extend(ai[4][0] for ai in addrinfos)
         except socket.gaierror as ex:
-            logging.error('%s [broker] Cannot resolve domain name %s (%s)', prepend_str, broker, str(ex))
+            logger(logging.ERROR, '[broker] Cannot resolve domain name %s (%s)', broker, str(ex))
 
-    logging.debug('%s [broker] Brokers resolved to %s', prepend_str, brokers_resolved)
+    logger(logging.DEBUG, '[broker] Brokers resolved to %s', brokers_resolved)
 
     if not brokers_resolved:
-        logging.fatal('%s [broker] No brokers resolved.', prepend_str)
+        logger(logging.FATAL, '[broker] No brokers resolved.')
         return None, None, None, None, None
 
     broker_timeout = 3
     if not broker_timeout:  # Allow zero in config
         broker_timeout = None
 
-    logging.info('%s [broker] Checking authentication method', prepend_str)
+    logger(logging.INFO, '[broker] Checking authentication method')
     use_ssl = True
     try:
         use_ssl = config_get_bool('messaging-hermes', 'use_ssl')
     except:
-        logging.info('%s [broker] Could not find use_ssl in configuration -- please update your rucio.cfg', prepend_str)
+        logger(logging.INFO, '[broker] Could not find use_ssl in configuration -- please update your rucio.cfg')
 
     port = config_get_int('messaging-hermes', 'port')
     vhost = config_get('messaging-hermes', 'broker_virtual_host', raise_exception=False)
@@ -132,13 +137,13 @@ def setup_activemq(prepend_str):
     conns = []
     for broker in brokers_resolved:
         if not use_ssl:
-            logging.info('%s [broker] setting up username/password authentication: %s', prepend_str, broker)
+            logger(logging.INFO, '[broker] setting up username/password authentication: %s', broker)
             con = stomp.Connection12(host_and_ports=[(broker, port)],
                                      vhost=vhost,
                                      keepalive=True,
                                      timeout=broker_timeout)
         else:
-            logging.info('%s [broker] setting up ssl cert/key authentication: %s', prepend_str, broker)
+            logger(logging.INFO, '[broker] setting up ssl cert/key authentication: %s', broker)
             con = stomp.Connection12(host_and_ports=[(broker, port)],
                                      use_ssl=True,
                                      ssl_key_file=config_get('messaging-hermes', 'ssl_key_file'),
@@ -155,17 +160,17 @@ def setup_activemq(prepend_str):
     return conns, destination, username, password, use_ssl
 
 
-def deliver_to_activemq(messages, conns, destination, username, password, use_ssl, prepend_str):
+def deliver_to_activemq(messages, conns, destination, username, password, use_ssl, logger):
     """
     Deliver messages to ActiveMQ
 
-    :param messages:     The list of messages.
-    :param conns:        A list of connections.
-    :param destination:  The destination topic or queue.
-    :param username:     The username if no SSL connection.
-    :param password:     The username if no SSL connection.
-    :param use_ssl:      Boolean to choose if SSL connection is used.
-    :param prepend_str:  The string to prepend to the logging.
+    :param messages:           The list of messages.
+    :param conns:              A list of connections.
+    :param destination:        The destination topic or queue.
+    :param username:           The username if no SSL connection.
+    :param password:           The username if no SSL connection.
+    :param use_ssl:            Boolean to choose if SSL connection is used.
+    :param logger:             The logger object.
     """
     to_delete = []
     for message in messages:
@@ -177,14 +182,14 @@ def deliver_to_activemq(messages, conns, destination, username, password, use_ss
                 labels = {'host': host_and_ports.split('.')[0]}
                 RECONNECT_COUNTER.labels(**labels).inc()
                 if not use_ssl:
-                    logging.info('%s [broker] - connecting with USERPASS to %s',
-                                 prepend_str,
-                                 host_and_ports)
+                    logger(logging.INFO,
+                           '[broker] - connecting with USERPASS to %s',
+                           host_and_ports)
                     conn.connect(username, password, wait=True)
                 else:
-                    logging.info('%s [broker] - connecting with SSL to %s',
-                                 prepend_str,
-                                 host_and_ports)
+                    logger(logging.INFO,
+                           '[broker] - connecting with SSL to %s',
+                           host_and_ports)
                     conn.connect(wait=True)
 
             conn.send(body=json.dumps({'event_type': str(message['event_type']).lower(),
@@ -196,73 +201,71 @@ def deliver_to_activemq(messages, conns, destination, username, password, use_ss
 
             to_delete.append(message['id'])
         except ValueError:
-            logging.error('%s [broker] Cannot serialize payload to JSON: %s',
-                          prepend_str,
-                          str(message['payload']))
+            logger(logging.ERROR,
+                   '[broker] Cannot serialize payload to JSON: %s',
+                   str(message['payload']))
             to_delete.append(message['id'])
             continue
         except stomp.exception.NotConnectedException as error:
-            logging.warning('%s [broker] Could not deliver message due to NotConnectedException: %s',
-                            prepend_str,
-                            str(error))
+            logger(logging.WARNING,
+                   '[broker] Could not deliver message due to NotConnectedException: %s',
+                   str(error))
             continue
         except stomp.exception.ConnectFailedException as error:
-            logging.warning('%s [broker] Could not deliver message due to ConnectFailedException: %s',
-                            prepend_str,
-                            str(error))
+            logger(logging.WARNING,
+                   '[broker] Could not deliver message due to ConnectFailedException: %s',
+                   str(error))
             continue
         except Exception as error:
-            logging.error('%s [broker] Could not deliver message: %s',
-                          prepend_str,
-                          str(error))
-            logging.critical(traceback.format_exc())
+            logger(logging.ERROR,
+                   '[broker] Could not deliver message: %s',
+                   str(error))
             continue
 
         if str(message['event_type']).lower().startswith('transfer') or str(message['event_type']).lower().startswith('stagein'):
-            logging.debug('%s [broker] - event_type: %s, scope: %s, name: %s, rse: %s, request-id: %s, transfer-id: %s, created_at: %s',
-                          prepend_str,
-                          str(message['event_type']).lower(),
-                          message['payload'].get('scope', None),
-                          message['payload'].get('name', None),
-                          message['payload'].get('dst-rse', None),
-                          message['payload'].get('request-id', None),
-                          message['payload'].get('transfer-id', None),
-                          str(message['created_at']))
+            logger(logging.DEBUG,
+                   '[broker] - event_type: %s, scope: %s, name: %s, rse: %s, request-id: %s, transfer-id: %s, created_at: %s',
+                   str(message['event_type']).lower(),
+                   message['payload'].get('scope', None),
+                   message['payload'].get('name', None),
+                   message['payload'].get('dst-rse', None),
+                   message['payload'].get('request-id', None),
+                   message['payload'].get('transfer-id', None),
+                   str(message['created_at']))
 
         elif str(message['event_type']).lower().startswith('dataset'):
-            logging.debug('%s [broker] - event_type: %s, scope: %s, name: %s, rse: %s, rule-id: %s, created_at: %s)',
-                          prepend_str,
-                          str(message['event_type']).lower(),
-                          message['payload']['scope'],
-                          message['payload']['name'],
-                          message['payload']['rse'],
-                          message['payload']['rule_id'],
-                          str(message['created_at']))
+            logger(logging.DEBUG, '[broker] - event_type: %s, scope: %s, name: %s, rse: %s, rule-id: %s, created_at: %s)',
+                   str(message['event_type']).lower(),
+                   message['payload']['scope'],
+                   message['payload']['name'],
+                   message['payload']['rse'],
+                   message['payload']['rule_id'],
+                   str(message['created_at']))
 
         elif str(message['event_type']).lower().startswith('deletion'):
             if 'url' not in message['payload']:
                 message['payload']['url'] = 'unknown'
-            logging.debug('%s [broker] - event_type: %s, scope: %s, name: %s, rse: %s, url: %s, created_at: %s)',
-                          prepend_str,
-                          str(message['event_type']).lower(),
-                          message['payload']['scope'],
-                          message['payload']['name'],
-                          message['payload']['rse'],
-                          message['payload']['url'],
-                          str(message['created_at']))
+            logger(logging.DEBUG,
+                   '[broker] - event_type: %s, scope: %s, name: %s, rse: %s, url: %s, created_at: %s)',
+                   str(message['event_type']).lower(),
+                   message['payload']['scope'],
+                   message['payload']['name'],
+                   message['payload']['rse'],
+                   message['payload']['url'],
+                   str(message['created_at']))
         else:
-            logging.debug('%s [broker] Other message: %s',
-                          prepend_str,
-                          message)
+            logger(logging.DEBUG,
+                   '[broker] Other message: %s',
+                   message)
     return to_delete
 
 
-def deliver_emails(messages, prepend_str):
+def deliver_emails(messages, logger):
     """
     Sends emails
 
-    :param messages:     The list of messages.
-    :param prepend_str:  The string to prepend to the logging.
+    :param messages:           The list of messages.
+    :param logger:             The logger object.
     """
 
     email_from = config_get('messaging-hermes', 'email_from')
@@ -289,20 +292,20 @@ def deliver_emails(messages, prepend_str):
                 smtp.quit()
                 to_delete.append(message['id'])
             except Exception as error:
-                logging.error('%s Cannot send email : %s', prepend_str, str(error))
+                logger(logging.ERROR, 'Cannot send email : %s', str(error))
         else:
             to_delete.append(message['id'])
             continue
     return to_delete
 
 
-def submit_to_elastic(messages, endpoint, prepend_str):
+def submit_to_elastic(messages, endpoint, logger):
     """
-    Aggregate a list of message to ElascticSearch
+    Aggregate a list of message to ElasticSearch
 
-    :param messages:     The list of messages.
-    :param endpoint:     The ES endpoint were to send the messages.
-    :param prepend_str:  The string to prepend to the logging.
+    :param messages:           The list of messages.
+    :param endpoint:           The ES endpoint were to send the messages.
+    :param logger:             The logger object.
     """
     text = ''
     for message in messages:
@@ -314,15 +317,15 @@ def submit_to_elastic(messages, endpoint, prepend_str):
     return res.status_code
 
 
-def aggregate_to_influx(messages, bin_size, endpoint, prepend_str):
+def aggregate_to_influx(messages, bin_size, endpoint, logger):
     """
     Aggregate a list of message using a certain bin_size
     and submit them to a InfluxDB endpoint
 
-    :param messages:     The list of messages.
-    :param bin_size:     The size of the bins for the aggreagation (e.g. 10m, 1h, etc.).
-    :param endpoint:     The InfluxDB endpoint were to send the messages.
-    :param prepend_str:  The string to prepend to the logging.
+    :param messages:           The list of messages.
+    :param bin_size:           The size of the bins for the aggreagation (e.g. 10m, 1h, etc.).
+    :param endpoint:           The InfluxDB endpoint were to send the messages.
+    :param logger:             The logger object.
     """
     bins = {}
     dtime = datetime.datetime.now()
@@ -335,6 +338,9 @@ def aggregate_to_influx(messages, bin_size, endpoint, prepend_str):
         event_type = message['event_type']
         payload = message['payload']
         if event_type in ['transfer-failed', 'transfer-done']:
+            if not payload['transferred_at']:
+                logger(logging.WARNING, 'No transferred_at for message. Reason : %s', payload['reason'])
+                continue
             transferred_at = time.strptime(payload['transferred_at'], '%Y-%m-%d %H:%M:%S')
             if bin_size == '1m':
                 transferred_at = int(calendar.timegm(transferred_at)) * 1000000000
@@ -389,7 +395,7 @@ def aggregate_to_influx(messages, bin_size, endpoint, prepend_str):
             points += '\\n'
     if points:
         res = requests.post(endpoint, data=points)
-        logging.debug('%s %s', prepend_str, str(res.text))
+        logger(logging.DEBUG, '%s', str(res.text))
         return res.status_code
     return 204
 
@@ -412,43 +418,44 @@ def hermes2(once=False, thread=0, bulk=1000, sleep_time=10):
     hb_thread = threading.current_thread()
     heartbeat.sanity_check(executable=executable, hostname=hostname, pid=pid, thread=hb_thread)
     heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)
-    prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'], heart_beat['nr_threads'])
 
     # Make an initial heartbeat so that all daemons have the correct worker number on the next try
     GRACEFUL_STOP.wait(10)
     heart_beat = heartbeat.live(executable, hostname, pid, hb_thread, older_than=3600)
+    prepend_str = 'hermes2[%i/%i] : ' % (heart_beat['assign_thread'], heart_beat['nr_threads'])
+    logger = formatted_logger(logging.log, prepend_str + '%s')
 
     try:
         services_list = get('hermes', 'services_list')
         services_list = services_list.split(',')
     except ConfigNotFound:
-        logging.debug('No services found, exiting')
+        logger(logging.DEBUG, 'No services found, exiting')
         sys.exit(1)
     if 'influx' in services_list:
         try:
             influx_endpoint = config_get('hermes', 'influxdb_endpoint', False, None)
             if not influx_endpoint:
-                logging.error('InfluxDB defined in the services list, but no endpoint can be find. Exiting')
+                logger(logging.ERROR, 'InfluxDB defined in the services list, but no endpoint can be find. Exiting')
                 sys.exit(1)
         except Exception as err:
-            logging.error(err)
+            logger(logging.ERROR, str(err))
     if 'elastic' in services_list:
         try:
             elastic_endpoint = config_get('hermes', 'elastic_endpoint', False, None)
             if not elastic_endpoint:
-                logging.error('Elastic defined in the services list, but no endpoint can be find. Exiting')
+                logger(logging.ERROR, 'Elastic defined in the services list, but no endpoint can be find. Exiting')
                 sys.exit(1)
         except Exception as err:
-            logging.error(err)
+            logger(logging.ERROR, str(err))
     if 'activemq' in services_list:
         try:
             # activemq_endpoint = config_get('hermes', 'activemq_endpoint', False, None)
-            conns, destination, username, password, use_ssl = setup_activemq(prepend_str)
+            conns, destination, username, password, use_ssl = setup_activemq(logger)
             if not conns:
-                logging.error('ActiveMQ defined in the services list, cannot be setup')
+                logger(logging.ERROR, 'ActiveMQ defined in the services list, cannot be setup')
                 sys.exit(1)
         except Exception as err:
-            logging.error(err)
+            logger(logging.ERROR, str(err))
 
     while not GRACEFUL_STOP.is_set():
         message_status = copy.deepcopy(services_list)
@@ -457,7 +464,8 @@ def hermes2(once=False, thread=0, bulk=1000, sleep_time=10):
         try:
             start_time = time.time()
             heart_beat = heartbeat.live(executable, hostname, pid, hb_thread, older_than=3600)
-            prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'], heart_beat['nr_threads'])
+            prepend_str = 'hermes2[%i/%i] : ' % (heart_beat['assign_thread'], heart_beat['nr_threads'])
+            logger = formatted_logger(logging.log, prepend_str + '%s')
             messages = retrieve_messages(bulk=bulk,
                                          thread=heart_beat['assign_thread'],
                                          total_threads=heart_beat['nr_threads'])
@@ -465,56 +473,56 @@ def hermes2(once=False, thread=0, bulk=1000, sleep_time=10):
             if messages:
                 for message in messages:
                     message_statuses[message['id']] = copy.deepcopy(services_list)
-                logging.debug('%s Retrieved %i messages retrieved in %s seconds', prepend_str, len(messages), time.time() - start_time)
+                logger(logging.DEBUG, 'Retrieved %i messages retrieved in %s seconds', len(messages), time.time() - start_time)
 
                 if 'influx' in message_status:
                     t_time = time.time()
-                    logging.debug('%s Will submit to influxDB', prepend_str)
+                    logger(logging.DEBUG, 'Will submit to influxDB')
                     try:
-                        state = aggregate_to_influx(messages=messages, bin_size='1m', endpoint=influx_endpoint, prepend_str=prepend_str)
+                        state = aggregate_to_influx(messages=messages, bin_size='1m', endpoint=influx_endpoint, logger=logger)
                     except Exception as error:
-                        logging.error('%s Error sending to InfluxDB : %s', prepend_str, str(error))
+                        logger(logging.ERROR, 'Error sending to InfluxDB : %s', str(error))
                         state = 500
                     if state in [204, 200]:
-                        logging.info('%s Messages successfully submitted to influxDB in %s seconds', prepend_str, time.time() - t_time)
+                        logger(logging.INFO, 'Messages successfully submitted to influxDB in %s seconds', time.time() - t_time)
                         for message in messages:
                             message_statuses[message['id']].remove('influx')
                     else:
-                        logging.info('%s Failure to submit to influxDB', prepend_str)
+                        logger(logging.INFO, 'Failure to submit to influxDB')
 
                 if 'elastic' in message_status:
                     t_time = time.time()
                     try:
-                        state = submit_to_elastic(messages=messages, endpoint=elastic_endpoint, prepend_str=prepend_str)
+                        state = submit_to_elastic(messages=messages, endpoint=elastic_endpoint, logger=logger)
                     except Exception as error:
-                        logging.error('%s Error sending to Elastic : %s', prepend_str, str(error))
+                        logger(logging.ERROR, 'Error sending to Elastic : %s', str(error))
                         state = 500
                     if state in [200, 204]:
-                        logging.info('%s Messages successfully submitted to elastic in %s seconds', prepend_str, time.time() - t_time)
+                        logger(logging.INFO, 'Messages successfully submitted to elastic in %s seconds', time.time() - t_time)
                         for message in messages:
                             message_statuses[message['id']].remove('elastic')
                     else:
-                        logging.info('%s Failure to submit to elastic', prepend_str)
+                        logger(logging.INFO, 'Failure to submit to elastic')
 
                 if 'emails' in message_status:
                     t_time = time.time()
                     try:
-                        to_delete = deliver_emails(messages=messages, prepend_str=prepend_str)
-                        logging.info('%s Messages successfully submitted by emails in %s seconds', prepend_str, time.time() - t_time)
+                        to_delete = deliver_emails(messages=messages, logger=logger)
+                        logger(logging.INFO, 'Messages successfully submitted by emails in %s seconds', time.time() - t_time)
                         for message_id in to_delete:
                             message_statuses[message_id].remove('emails')
                     except Exception as error:
-                        logging.error('%s Error sending email : %s', prepend_str, str(error))
+                        logger(logging.ERROR, 'Error sending email : %s', str(error))
 
                 if 'activemq' in message_status:
                     t_time = time.time()
                     try:
-                        to_delete = deliver_to_activemq(messages=messages, conns=conns, destination=destination, username=username, password=password, use_ssl=use_ssl, prepend_str=prepend_str)
-                        logging.info('%s Messages successfully submitted to ActiveMQ in %s seconds', prepend_str, time.time() - t_time)
+                        to_delete = deliver_to_activemq(messages=messages, conns=conns, destination=destination, username=username, password=password, use_ssl=use_ssl, logger=logger)
+                        logger(logging.INFO, 'Messages successfully submitted to ActiveMQ in %s seconds', time.time() - t_time)
                         for message_id in to_delete:
                             message_statuses[message_id].remove('activemq')
                     except Exception as error:
-                        logging.error('%s Error sending to ActiveMQ : %s', prepend_str, str(error))
+                        logger(logging.ERROR, 'Error sending to ActiveMQ : %s', str(error))
 
                 to_delete = []
                 to_update = {}
@@ -535,21 +543,21 @@ def hermes2(once=False, thread=0, bulk=1000, sleep_time=10):
                                                   'updated_at': message['created_at'],
                                                   'payload': str(message['payload']),
                                                   'event_type': message['event_type']})
-                logging.info('%s Deleting %s messages', prepend_str, len(to_delete))
+                logger(logging.INFO, 'Deleting %s messages', len(to_delete))
                 delete_messages(messages=to_delete)
                 for status in to_update:
-                    logging.info('%s Failure to submit %s messages to %s. Will update the message status', prepend_str, str(len(to_update[status])), status)
+                    logger(logging.INFO, 'Failure to submit %s messages to %s. Will update the message status', str(len(to_update[status])), status)
                     update_messages_services(messages=to_update[status], services=status)
 
             if once:
                 break
             tottime = time.time() - stime
             if tottime < sleep_time:
-                logging.info('%s Will sleep for %s seconds', prepend_str, sleep_time - tottime)
+                logger(logging.INFO, 'Will sleep for %s seconds', sleep_time - tottime)
                 time.sleep(sleep_time - tottime)
 
-        except:
-            logging.critical(traceback.format_exc())
+        except Exception:
+            logger(logging.ERROR, "Failed to submit messages", exc_info=True)
 
 
 def stop(signum=None, frame=None):