Repository: rucio/rucio
Issue: VO missing from reaper messages

Motivation
----------

Hermes messages for transfers etc. include the VO as part of the payload when it is not `def` (i.e. in M-VO mode only). This allows messages to be sorted by VO when monitoring. Currently, none of the reapers include the VO as part of the message payload.

Modification
------------

Use the scope of the deleted replica to get the VO, and include it in the payload for all reaper daemons.


Changed files:
lib/rucio/daemons/reaper/light_reaper.py, lib/rucio/daemons/reaper/reaper2.py, lib/rucio/daemons/reaper/dark_reaper.py, lib/rucio/daemons/reaper/reaper.py

diff --git a/lib/rucio/daemons/reaper/dark_reaper.py b/lib/rucio/daemons/reaper/dark_reaper.py
index ef1af12b0..744ebdd69 100644
--- a/lib/rucio/daemons/reaper/dark_reaper.py
+++ b/lib/rucio/daemons/reaper/dark_reaper.py
@@ -118,15 +118,18 @@ def reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False
                             prot.delete(pfn)
                             duration = time.time() - start
                             logging.info('Dark Reaper %s-%s: Deletion SUCCESS of %s:%s as %s on %s in %s seconds', worker_number, total_workers, scope, replica['name'], pfn, rse, duration)
-                            add_message('deletion-done', {'scope': scope,
-                                                          'name': replica['name'],
-                                                          'rse': rse,
-                                                          'rse_id': rse_id,
-                                                          'file-size': replica.get('bytes') or 0,
-                                                          'bytes': replica.get('bytes') or 0,
-                                                          'url': pfn,
-                                                          'duration': duration,
-                                                          'protocol': prot.attributes['scheme']})
+                            payload = {'scope': scope,
+                                       'name': replica['name'],
+                                       'rse': rse,
+                                       'rse_id': rse_id,
+                                       'file-size': replica.get('bytes') or 0,
+                                       'bytes': replica.get('bytes') or 0,
+                                       'url': pfn,
+                                       'duration': duration,
+                                       'protocol': prot.attributes['scheme']}
+                            if replica['scope'].vo != 'def':
+                                payload['vo'] = replica['scope'].vo
+                            add_message('deletion-done', payload)
                             deleted_replicas.append(replica)
                         except SourceNotFound:
                             err_msg = 'Dark Reaper %s-%s: Deletion NOTFOUND of %s:%s as %s on %s' % (worker_number, total_workers, scope, replica['name'], pfn, rse)
@@ -135,15 +138,18 @@ def reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False
                         except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:
                             err_msg = 'Dark Reaper %s-%s: Deletion NOACCESS of %s:%s as %s on %s: %s' % (worker_number, total_workers, scope, replica['name'], pfn, rse, str(error))
                             logging.warning(err_msg)
-                            add_message('deletion-failed', {'scope': scope,
-                                                            'name': replica['name'],
-                                                            'rse': rse,
-                                                            'rse_id': rse_id,
-                                                            'file-size': replica['bytes'] or 0,
-                                                            'bytes': replica['bytes'] or 0,
-                                                            'url': pfn,
-                                                            'reason': str(error),
-                                                            'protocol': prot.attributes['scheme']})
+                            payload = {'scope': scope,
+                                       'name': replica['name'],
+                                       'rse': rse,
+                                       'rse_id': rse_id,
+                                       'file-size': replica['bytes'] or 0,
+                                       'bytes': replica['bytes'] or 0,
+                                       'url': pfn,
+                                       'reason': str(error),
+                                       'protocol': prot.attributes['scheme']}
+                            if replica['scope'].vo != 'def':
+                                payload['vo'] = replica['scope'].vo
+                            add_message('deletion-failed', payload)
 
                         except Exception:
                             logging.critical(traceback.format_exc())
diff --git a/lib/rucio/daemons/reaper/light_reaper.py b/lib/rucio/daemons/reaper/light_reaper.py
index 9ac0841f0..15daccf86 100644
--- a/lib/rucio/daemons/reaper/light_reaper.py
+++ b/lib/rucio/daemons/reaper/light_reaper.py
@@ -113,15 +113,18 @@ def reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False
                             prot.delete(pfn)
                             duration = time.time() - start
                             logging.info('Light Reaper %s-%s: Deletion SUCCESS of %s:%s as %s on %s in %s seconds', worker_number, total_workers, replica['scope'], replica['name'], pfn, rse, duration)
-                            add_message('deletion-done', {'scope': replica['scope'].external,
-                                                          'name': replica['name'],
-                                                          'rse': rse,
-                                                          'rse_id': rse_id,
-                                                          'file-size': replica.get('bytes') or 0,
-                                                          'bytes': replica.get('bytes') or 0,
-                                                          'url': pfn,
-                                                          'duration': duration,
-                                                          'protocol': prot.attributes['scheme']})
+                            payload = {'scope': replica['scope'].external,
+                                       'name': replica['name'],
+                                       'rse': rse,
+                                       'rse_id': rse_id,
+                                       'file-size': replica.get('bytes') or 0,
+                                       'bytes': replica.get('bytes') or 0,
+                                       'url': pfn,
+                                       'duration': duration,
+                                       'protocol': prot.attributes['scheme']}
+                            if replica['scope'].vo != 'def':
+                                payload['vo'] = replica['scope'].vo
+                            add_message('deletion-done', payload)
                             deleted_replicas.append(replica)
                         except SourceNotFound:
                             err_msg = 'Light Reaper %s-%s: Deletion NOTFOUND of %s:%s as %s on %s' % (worker_number, total_workers, replica['scope'], replica['name'], pfn, rse)
@@ -130,15 +133,18 @@ def reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False
                         except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:
                             err_msg = 'Light Reaper %s-%s: Deletion NOACCESS of %s:%s as %s on %s: %s' % (worker_number, total_workers, replica['scope'], replica['name'], pfn, rse, str(error))
                             logging.warning(err_msg)
-                            add_message('deletion-failed', {'scope': replica['scope'].external,
-                                                            'name': replica['name'],
-                                                            'rse': rse,
-                                                            'rse_id': rse_id,
-                                                            'file-size': replica['bytes'] or 0,
-                                                            'bytes': replica['bytes'] or 0,
-                                                            'url': pfn,
-                                                            'reason': str(error),
-                                                            'protocol': prot.attributes['scheme']})
+                            payload = {'scope': replica['scope'].external,
+                                       'name': replica['name'],
+                                       'rse': rse,
+                                       'rse_id': rse_id,
+                                       'file-size': replica['bytes'] or 0,
+                                       'bytes': replica['bytes'] or 0,
+                                       'url': pfn,
+                                       'reason': str(error),
+                                       'protocol': prot.attributes['scheme']}
+                            if replica['scope'].vo != 'def':
+                                payload['vo'] = replica['scope'].vo
+                            add_message('deletion-failed', payload)
 
                         except:
                             logging.critical(traceback.format_exc())
diff --git a/lib/rucio/daemons/reaper/reaper.py b/lib/rucio/daemons/reaper/reaper.py
index 88eeed0c2..c14d86823 100644
--- a/lib/rucio/daemons/reaper/reaper.py
+++ b/lib/rucio/daemons/reaper/reaper.py
@@ -241,6 +241,16 @@ def reaper(rses, worker_number=0, child_number=0, total_children=1, chunk_size=1
                                 prot.connect()
                                 for replica in files:
                                     try:
+                                        deletion_dict = {'scope': replica['scope'].external,
+                                                         'name': replica['name'],
+                                                         'rse': rse_info['rse'],
+                                                         'rse_id': rse_info['id'],
+                                                         'file-size': replica['bytes'],
+                                                         'bytes': replica['bytes'],
+                                                         'url': replica['pfn'],
+                                                         'protocol': prot.attributes['scheme']}
+                                        if replica['scope'].vo != 'def':
+                                            deletion_dict['vo'] = replica['scope'].vo
                                         logging.info('Reaper %s-%s: Deletion ATTEMPT of %s:%s as %s on %s', worker_number, child_number, replica['scope'], replica['name'], replica['pfn'], rse['rse'])
                                         start = time.time()
                                         if rse['staging_area'] or rse['rse'].endswith("STAGING"):
@@ -260,66 +270,41 @@ def reaper(rses, worker_number=0, child_number=0, total_children=1, chunk_size=1
 
                                         deleted_files.append({'scope': replica['scope'], 'name': replica['name']})
 
-                                        add_message('deletion-done', {'scope': replica['scope'].external,
-                                                                      'name': replica['name'],
-                                                                      'rse': rse_info['rse'],
-                                                                      'rse_id': rse_info['id'],
-                                                                      'file-size': replica['bytes'],
-                                                                      'bytes': replica['bytes'],
-                                                                      'url': replica['pfn'],
-                                                                      'duration': duration,
-                                                                      'protocol': prot.attributes['scheme']})
+                                        deletion_dict['duration'] = duration
+                                        add_message('deletion-done', deletion_dict)
                                         logging.info('Reaper %s-%s: Deletion SUCCESS of %s:%s as %s on %s in %s seconds', worker_number, child_number, replica['scope'], replica['name'], replica['pfn'], rse['rse'], duration)
                                     except SourceNotFound:
                                         err_msg = 'Deletion NOTFOUND of %s:%s as %s on %s' % (replica['scope'], replica['name'], replica['pfn'], rse['rse'])
                                         logging.warning(err_msg)
                                         deleted_files.append({'scope': replica['scope'], 'name': replica['name']})
                                         if replica['state'] == ReplicaState.AVAILABLE:
-                                            add_message('deletion-failed', {'scope': replica['scope'].external,
-                                                                            'name': replica['name'],
-                                                                            'rse': rse_info['rse'],
-                                                                            'rse_id': rse_info['id'],
-                                                                            'file-size': replica['bytes'],
-                                                                            'bytes': replica['bytes'],
-                                                                            'url': replica['pfn'],
-                                                                            'reason': str(err_msg),
-                                                                            'protocol': prot.attributes['scheme']})
+                                            deletion_dict['reason'] = str(err_msg)
+                                            add_message('deletion-failed', deletion_dict)
                                     except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:
                                         logging.warning('Reaper %s-%s: Deletion NOACCESS of %s:%s as %s on %s: %s', worker_number, child_number, replica['scope'], replica['name'], replica['pfn'], rse['rse'], str(error))
-                                        add_message('deletion-failed', {'scope': replica['scope'].external,
-                                                                        'name': replica['name'],
-                                                                        'rse': rse_info['rse'],
-                                                                        'rse_id': rse_info['id'],
-                                                                        'file-size': replica['bytes'],
-                                                                        'bytes': replica['bytes'],
-                                                                        'url': replica['pfn'],
-                                                                        'reason': str(error),
-                                                                        'protocol': prot.attributes['scheme']})
+                                        deletion_dict['reason'] = str(error)
+                                        add_message('deletion-failed', deletion_dict)
                                     except Exception as error:
                                         logging.critical('Reaper %s-%s: Deletion CRITICAL of %s:%s as %s on %s: %s', worker_number, child_number, replica['scope'], replica['name'], replica['pfn'], rse['rse'], str(traceback.format_exc()))
-                                        add_message('deletion-failed', {'scope': replica['scope'].external,
-                                                                        'name': replica['name'],
-                                                                        'rse': rse_info['rse'],
-                                                                        'rse_id': rse_info['id'],
-                                                                        'file-size': replica['bytes'],
-                                                                        'bytes': replica['bytes'],
-                                                                        'url': replica['pfn'],
-                                                                        'reason': str(error),
-                                                                        'protocol': prot.attributes['scheme']})
+                                        deletion_dict['reason'] = str(error)
+                                        add_message('deletion-failed', deletion_dict)
                                     except:
                                         logging.critical('Reaper %s-%s: Deletion CRITICAL of %s:%s as %s on %s: %s', worker_number, child_number, replica['scope'], replica['name'], replica['pfn'], rse['rse'], str(traceback.format_exc()))
                             except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:
                                 for replica in files:
                                     logging.warning('Reaper %s-%s: Deletion NOACCESS of %s:%s as %s on %s: %s', worker_number, child_number, replica['scope'], replica['name'], replica['pfn'], rse['rse'], str(error))
-                                    add_message('deletion-failed', {'scope': replica['scope'].external,
-                                                                    'name': replica['name'],
-                                                                    'rse': rse_info['rse'],
-                                                                    'rse_id': rse_info['id'],
-                                                                    'file-size': replica['bytes'],
-                                                                    'bytes': replica['bytes'],
-                                                                    'url': replica['pfn'],
-                                                                    'reason': str(error),
-                                                                    'protocol': prot.attributes['scheme']})
+                                    payload = {'scope': replica['scope'].external,
+                                               'name': replica['name'],
+                                               'rse': rse_info['rse'],
+                                               'rse_id': rse_info['id'],
+                                               'file-size': replica['bytes'],
+                                               'bytes': replica['bytes'],
+                                               'url': replica['pfn'],
+                                               'reason': str(error),
+                                               'protocol': prot.attributes['scheme']}
+                                    if replica['scope'].vo != 'def':
+                                        deletion_dict['vo'] = replica['scope'].vo
+                                    add_message('deletion-failed', payload)
                                     break
                             finally:
                                 prot.close()
diff --git a/lib/rucio/daemons/reaper/reaper2.py b/lib/rucio/daemons/reaper/reaper2.py
index aae8d7b78..e909a92ea 100644
--- a/lib/rucio/daemons/reaper/reaper2.py
+++ b/lib/rucio/daemons/reaper/reaper2.py
@@ -170,6 +170,8 @@ def delete_from_storage(replicas, prot, rse_info, staging_areas, prepend_str):
                                  'bytes': replica['bytes'],
                                  'url': replica['pfn'],
                                  'protocol': prot.attributes['scheme']}
+                if replica['scope'].vo != 'def':
+                    deletion_dict['vo'] = replica['scope'].vo
                 logging.info('%s Deletion ATTEMPT of %s:%s as %s on %s', prepend_str, replica['scope'], replica['name'], replica['pfn'], rse_name)
                 start = time.time()
                 # For STAGING RSEs, no physical deletion
@@ -214,14 +216,17 @@ def delete_from_storage(replicas, prot, rse_info, staging_areas, prepend_str):
     except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:
         for replica in replicas:
             logging.warning('%s Deletion NOACCESS of %s:%s as %s on %s: %s', prepend_str, replica['scope'], replica['name'], replica['pfn'], rse_name, str(error))
-            add_message('deletion-failed', {'scope': replica['scope'].external,
-                                            'name': replica['name'],
-                                            'rse': rse_name,
-                                            'file-size': replica['bytes'],
-                                            'bytes': replica['bytes'],
-                                            'url': replica['pfn'],
-                                            'reason': str(error),
-                                            'protocol': prot.attributes['scheme']})
+            payload = {'scope': replica['scope'].external,
+                       'name': replica['name'],
+                       'rse': rse_name,
+                       'file-size': replica['bytes'],
+                       'bytes': replica['bytes'],
+                       'url': replica['pfn'],
+                       'reason': str(error),
+                       'protocol': prot.attributes['scheme']}
+            if replica['scope'].vo != 'def':
+                payload['vo'] = replica['scope'].vo
+            add_message('deletion-failed', payload)
 
     finally:
         prot.close()