We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Dark Reaper does not work with Python 3
Motivation
----------

```
Traceback (most recent call last):
  File "/usr/lib64/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/lib64/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.6/site-packages/rucio/daemons/reaper/dark_reaper.py", line 84, in reaper
    hash_executable = hashlib.sha256(sys.argv[0] + ''.join(rses)).hexdigest()
TypeError: Unicode-objects must be encoded before hashing
```

```
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/rucio/daemons/reaper/dark_reaper.py", line 114, in reaper
    operation='delete', scheme=scheme).values()[0])
TypeError: 'dict_values' object does not support indexing
```

Modification
------------

Port to Python 3.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lib/rucio/daemons/reaper/dark_reaper.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # Copyright 2016-2020 CERN
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #    http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 # Authors:
17 # - Vincent Garonne <vincent.garonne@cern.ch>, 2016-2018
18 # - Martin Barisits <martin.barisits@cern.ch>, 2016
19 # - Thomas Beermann <thomas.beermann@cern.ch>, 2016-2019
20 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019
21 # - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019
22 # - Cedric Serfon <cedric.serfon@cern.ch>, 2020
23 # - Brandon White <bjwhite@fnal.gov>, 2019
24 # - Patrick Austin <patrick.austin@stfc.ac.uk>, 2020
25 # - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020
26 # - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020
27 
28 '''
29 Dark Reaper is a daemon to manage quarantined file deletion.
30 '''
31 
32 import hashlib
33 import logging
34 import os
35 import random
36 import socket
37 import sys
38 import threading
39 import time
40 import traceback
41 
42 import rucio.db.sqla.util
43 from rucio.common import exception
44 from rucio.common.config import config_get, config_get_bool
45 from rucio.common.exception import (SourceNotFound, DatabaseException, ServiceUnavailable,
46                                     RSEAccessDenied, ResourceTemporaryUnavailable,
47                                     RSENotFound, VONotFound)
48 from rucio.core.heartbeat import live, die, sanity_check
49 from rucio.core.message import add_message
50 from rucio.core.quarantined_replica import (list_quarantined_replicas,
51                                             delete_quarantined_replicas,
52                                             list_rses)
53 from rucio.core.rse_expression_parser import parse_expression
54 from rucio.core.vo import list_vos
55 from rucio.rse import rsemanager as rsemgr
56 
57 logging.getLogger("requests").setLevel(logging.CRITICAL)
58 
59 logging.basicConfig(stream=sys.stdout,
60                     level=getattr(logging,
61                                   config_get('common', 'loglevel',
62                                              raise_exception=False,
63                                              default='DEBUG').upper()),
64                     format='%(asctime)s\t%(process)d\t%(levelname)s\t%(message)s')
65 
66 GRACEFUL_STOP = threading.Event()
67 
68 
69 def reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False, scheme=None):
70     """
71     Main loop to select and delete files.
72 
73     :param rses: List of RSEs the reaper should work against. If empty, it considers all RSEs.
74     :param worker_number: The worker number.
75     :param total_workers:  The total number of workers.
76     :param chunk_size: the size of chunk for deletion.
77     :param once: If True, only runs one iteration of the main loop.
78     :param scheme: Force the reaper to use a particular protocol, e.g., mock.
79     """
80     logging.info('Starting Dark Reaper %s-%s: Will work on RSEs: %s', worker_number, total_workers, ', '.join(rses))
81 
82     pid = os.getpid()
83     thread = threading.current_thread()
84     hostname = socket.gethostname()
85     executable = ' '.join(sys.argv)
86     hash_executable = hashlib.sha256(sys.argv[0] + ''.join(rses)).hexdigest()
87     sanity_check(executable=None, hostname=hostname)
88 
89     while not GRACEFUL_STOP.is_set():
90         try:
91             # heartbeat
92             heartbeat = live(executable=executable, hostname=hostname, pid=pid, thread=thread, hash_executable=hash_executable)
93             logging.info('Dark Reaper({0[worker_number]}/{0[total_workers]}): Live gives {0[heartbeat]}'.format(locals()))
94             nothing_to_do = True
95 
96             random.shuffle(rses)
97             for rse_id in rses:
98                 replicas = list_quarantined_replicas(rse_id=rse_id,
99                                                      limit=chunk_size, worker_number=worker_number,
100                                                      total_workers=total_workers)
101 
102                 rse_info = rsemgr.get_rse_info(rse_id=rse_id)
103                 rse = rse_info['rse']
104                 prot = rsemgr.create_protocol(rse_info, 'delete', scheme=scheme)
105                 deleted_replicas = []
106                 try:
107                     prot.connect()
108                     for replica in replicas:
109                         nothing_to_do = False
110                         scope = ''
111                         if replica['scope']:
112                             scope = replica['scope'].external
113                         try:
114                             pfn = str(rsemgr.lfns2pfns(rse_settings=rse_info,
115                                                        lfns=[{'scope': scope, 'name': replica['name'], 'path': replica['path']}],
116                                                        operation='delete', scheme=scheme).values()[0])
117                             logging.info('Dark Reaper %s-%s: Deletion ATTEMPT of %s:%s as %s on %s', worker_number, total_workers, scope, replica['name'], pfn, rse)
118                             start = time.time()
119                             prot.delete(pfn)
120                             duration = time.time() - start
121                             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)
122                             payload = {'scope': scope,
123                                        'name': replica['name'],
124                                        'rse': rse,
125                                        'rse_id': rse_id,
126                                        'file-size': replica.get('bytes') or 0,
127                                        'bytes': replica.get('bytes') or 0,
128                                        'url': pfn,
129                                        'duration': duration,
130                                        'protocol': prot.attributes['scheme']}
131                             if replica['scope'].vo != 'def':
132                                 payload['vo'] = replica['scope'].vo
133                             add_message('deletion-done', payload)
134                             deleted_replicas.append(replica)
135                         except SourceNotFound:
136                             err_msg = 'Dark Reaper %s-%s: Deletion NOTFOUND of %s:%s as %s on %s' % (worker_number, total_workers, scope, replica['name'], pfn, rse)
137                             logging.warning(err_msg)
138                             deleted_replicas.append(replica)
139                         except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:
140                             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))
141                             logging.warning(err_msg)
142                             payload = {'scope': scope,
143                                        'name': replica['name'],
144                                        'rse': rse,
145                                        'rse_id': rse_id,
146                                        'file-size': replica['bytes'] or 0,
147                                        'bytes': replica['bytes'] or 0,
148                                        'url': pfn,
149                                        'reason': str(error),
150                                        'protocol': prot.attributes['scheme']}
151                             if replica['scope'].vo != 'def':
152                                 payload['vo'] = replica['scope'].vo
153                             add_message('deletion-failed', payload)
154 
155                         except Exception:
156                             logging.critical(traceback.format_exc())
157                 finally:
158                     prot.close()
159 
160                 delete_quarantined_replicas(rse_id=rse_id, replicas=deleted_replicas)
161 
162                 if once:
163                     break
164 
165             if once:
166                 break
167 
168             if nothing_to_do:
169                 logging.info('Dark Reaper %s-%s: Nothing to do. I will sleep for 60s', worker_number, total_workers)
170                 time.sleep(60)
171 
172         except DatabaseException as error:
173             logging.warning('Reaper:  %s', str(error))
174         except Exception:
175             logging.critical(traceback.format_exc())
176 
177     die(executable=executable, hostname=hostname, pid=pid, thread=thread, hash_executable=hash_executable)
178     logging.info('Graceful stop requested')
179     logging.info('Graceful stop done')
180     return
181 
182 
183 def stop(signum=None, frame=None):
184     """
185     Graceful exit.
186     """
187     GRACEFUL_STOP.set()
188 
189 
190 def run(total_workers=1, chunk_size=100, once=False, rses=[], scheme=None,
191         exclude_rses=None, include_rses=None, vos=None, delay_seconds=0):
192     """
193     Starts up the reaper threads.
194 
195     :param total_workers: The total number of workers.
196     :param chunk_size: the size of chunk for deletion.
197     :param once: If True, only runs one iteration of the main loop.
198     :param rses: List of RSEs the reaper should work against. If empty, it considers all RSEs (Single-VO only).
199     :param scheme: Force the reaper to use a particular protocol/scheme, e.g., mock.
200     :param exclude_rses: RSE expression to exclude RSEs from the Reaper.
201     :param include_rses: RSE expression to include RSEs.
202     :param vos: VOs on which to look for RSEs. Only used in multi-VO mode.
203                 If None, we either use all VOs if run from "def", or the current VO otherwise.
204     """
205     if rucio.db.sqla.util.is_old_db():
206         raise exception.DatabaseException('Database was not updated, daemon won\'t start')
207 
208     logging.info('main: starting processes')
209 
210     multi_vo = config_get_bool('common', 'multi_vo', raise_exception=False, default=False)
211     if not multi_vo:
212         if vos:
213             logging.warning('Ignoring argument vos, this is only applicable in a multi-VO setup.')
214         vos = ['def']
215     else:
216         if vos:
217             invalid = set(vos) - set([v['vo'] for v in list_vos()])
218             if invalid:
219                 msg = 'VO{} {} cannot be found'.format('s' if len(invalid) > 1 else '', ', '.join([repr(v) for v in invalid]))
220                 raise VONotFound(msg)
221         else:
222             vos = [v['vo'] for v in list_vos()]
223         logging.info('Dark Reaper: This instance will work on VO%s: %s' % ('s' if len(vos) > 1 else '', ', '.join([v for v in vos])))
224 
225     all_rses = []
226     for vo in vos:
227         all_rses.extend(list_rses(filters={'vo': vo}))
228 
229     if rses:
230         invalid = set(rses) - set([rse['rse'] for rse in all_rses])
231         if invalid:
232             msg = 'RSE{} {} cannot be found'.format('s' if len(invalid) > 1 else '',
233                                                     ', '.join([repr(rse) for rse in invalid]))
234             raise RSENotFound(msg)
235         rses = [rse for rse in all_rses if rse['rse'] in rses]
236     else:
237         rses = all_rses
238 
239     if exclude_rses:
240         excluded_rses = parse_expression(exclude_rses)
241         rses = [rse for rse in rses if rse not in excluded_rses]
242 
243     if include_rses:
244         included_rses = parse_expression(include_rses)
245         rses = [rse for rse in rses if rse in included_rses]
246 
247     if not rses:
248         logging.error('Dark Reaper: No RSEs found. Exiting.')
249         return
250 
251     threads = []
252     for worker in range(total_workers):
253         kwargs = {'worker_number': worker,
254                   'total_workers': total_workers,
255                   'rses': rses,
256                   'once': once,
257                   'chunk_size': chunk_size,
258                   'scheme': scheme}
259         threads.append(threading.Thread(target=reaper, kwargs=kwargs, name='Worker: %s, Total_Workers: %s' % (worker, total_workers)))
260     [t.start() for t in threads]
261     while threads[0].is_alive():
262         [t.join(timeout=3.14) for t in threads]
263
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lib/rucio/daemons/reaper/dark_reaper.py b/lib/rucio/daemons/reaper/dark_reaper.py
--- a/lib/rucio/daemons/reaper/dark_reaper.py
+++ b/lib/rucio/daemons/reaper/dark_reaper.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2016-2020 CERN
+# Copyright 2016-2021 CERN
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -23,7 +23,7 @@
 # - Brandon White <bjwhite@fnal.gov>, 2019
 # - Patrick Austin <patrick.austin@stfc.ac.uk>, 2020
 # - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020
-# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020-2021
 
 '''
 Dark Reaper is a daemon to manage quarantined file deletion.
@@ -83,7 +83,7 @@
     thread = threading.current_thread()
     hostname = socket.gethostname()
     executable = ' '.join(sys.argv)
-    hash_executable = hashlib.sha256(sys.argv[0] + ''.join(rses)).hexdigest()
+    hash_executable = hashlib.sha256((sys.argv[0] + ''.join(rses)).encode()).hexdigest()
     sanity_check(executable=None, hostname=hostname)
 
     while not GRACEFUL_STOP.is_set():
@@ -111,9 +111,12 @@
                         if replica['scope']:
                             scope = replica['scope'].external
                         try:
-                            pfn = str(rsemgr.lfns2pfns(rse_settings=rse_info,
-                                                       lfns=[{'scope': scope, 'name': replica['name'], 'path': replica['path']}],
-                                                       operation='delete', scheme=scheme).values()[0])
+                            pfn = str(list(rsemgr.lfns2pfns(rse_settings=rse_info,
+                                                            lfns=[{'scope': scope,
+                                                                   'name': replica['name'],
+                                                                   'path': replica['path']}],
+                                                            operation='delete',
+                                                            scheme=scheme).values())[0])
                             logging.info('Dark Reaper %s-%s: Deletion ATTEMPT of %s:%s as %s on %s', worker_number, total_workers, scope, replica['name'], pfn, rse)
                             start = time.time()
                             prot.delete(pfn)

{"golden_diff": "diff --git a/lib/rucio/daemons/reaper/dark_reaper.py b/lib/rucio/daemons/reaper/dark_reaper.py\n--- a/lib/rucio/daemons/reaper/dark_reaper.py\n+++ b/lib/rucio/daemons/reaper/dark_reaper.py\n@@ -1,5 +1,5 @@\n # -*- coding: utf-8 -*-\n-# Copyright 2016-2020 CERN\n+# Copyright 2016-2021 CERN\n #\n # Licensed under the Apache License, Version 2.0 (the \"License\");\n # you may not use this file except in compliance with the License.\n@@ -23,7 +23,7 @@\n # - Brandon White <bjwhite@fnal.gov>, 2019\n # - Patrick Austin <patrick.austin@stfc.ac.uk>, 2020\n # - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020\n-# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020\n+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020-2021\n \n '''\n Dark Reaper is a daemon to manage quarantined file deletion.\n@@ -83,7 +83,7 @@\n     thread = threading.current_thread()\n     hostname = socket.gethostname()\n     executable = ' '.join(sys.argv)\n-    hash_executable = hashlib.sha256(sys.argv[0] + ''.join(rses)).hexdigest()\n+    hash_executable = hashlib.sha256((sys.argv[0] + ''.join(rses)).encode()).hexdigest()\n     sanity_check(executable=None, hostname=hostname)\n \n     while not GRACEFUL_STOP.is_set():\n@@ -111,9 +111,12 @@\n                         if replica['scope']:\n                             scope = replica['scope'].external\n                         try:\n-                            pfn = str(rsemgr.lfns2pfns(rse_settings=rse_info,\n-                                                       lfns=[{'scope': scope, 'name': replica['name'], 'path': replica['path']}],\n-                                                       operation='delete', scheme=scheme).values()[0])\n+                            pfn = str(list(rsemgr.lfns2pfns(rse_settings=rse_info,\n+                                                            lfns=[{'scope': scope,\n+                                                                   'name': replica['name'],\n+                                                                   'path': replica['path']}],\n+                                                            operation='delete',\n+                                                            scheme=scheme).values())[0])\n                             logging.info('Dark Reaper %s-%s: Deletion ATTEMPT of %s:%s as %s on %s', worker_number, total_workers, scope, replica['name'], pfn, rse)\n                             start = time.time()\n                             prot.delete(pfn)\n", "issue": "Dark Reaper does not work with Python 3\nMotivation\r\n----------\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/usr/lib64/python3.6/threading.py\", line 916, in _bootstrap_inner\r\n    self.run()\r\n  File \"/usr/lib64/python3.6/threading.py\", line 864, in run\r\n    self._target(*self._args, **self._kwargs)\r\n  File \"/usr/local/lib/python3.6/site-packages/rucio/daemons/reaper/dark_reaper.py\", line 84, in reaper\r\n    hash_executable = hashlib.sha256(sys.argv[0] + ''.join(rses)).hexdigest()\r\nTypeError: Unicode-objects must be encoded before hashing\r\n```\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.6/site-packages/rucio/daemons/reaper/dark_reaper.py\", line 114, in reaper\r\n    operation='delete', scheme=scheme).values()[0])\r\nTypeError: 'dict_values' object does not support indexing\r\n```\r\n\r\nModification\r\n------------\r\n\r\nPort to Python 3.\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2016-2020 CERN\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n# Authors:\n# - Vincent Garonne <vincent.garonne@cern.ch>, 2016-2018\n# - Martin Barisits <martin.barisits@cern.ch>, 2016\n# - Thomas Beermann <thomas.beermann@cern.ch>, 2016-2019\n# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019\n# - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019\n# - Cedric Serfon <cedric.serfon@cern.ch>, 2020\n# - Brandon White <bjwhite@fnal.gov>, 2019\n# - Patrick Austin <patrick.austin@stfc.ac.uk>, 2020\n# - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020\n# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020\n\n'''\nDark Reaper is a daemon to manage quarantined file deletion.\n'''\n\nimport hashlib\nimport logging\nimport os\nimport random\nimport socket\nimport sys\nimport threading\nimport time\nimport traceback\n\nimport rucio.db.sqla.util\nfrom rucio.common import exception\nfrom rucio.common.config import config_get, config_get_bool\nfrom rucio.common.exception import (SourceNotFound, DatabaseException, ServiceUnavailable,\n                                    RSEAccessDenied, ResourceTemporaryUnavailable,\n                                    RSENotFound, VONotFound)\nfrom rucio.core.heartbeat import live, die, sanity_check\nfrom rucio.core.message import add_message\nfrom rucio.core.quarantined_replica import (list_quarantined_replicas,\n                                            delete_quarantined_replicas,\n                                            list_rses)\nfrom rucio.core.rse_expression_parser import parse_expression\nfrom rucio.core.vo import list_vos\nfrom rucio.rse import rsemanager as rsemgr\n\nlogging.getLogger(\"requests\").setLevel(logging.CRITICAL)\n\nlogging.basicConfig(stream=sys.stdout,\n                    level=getattr(logging,\n                                  config_get('common', 'loglevel',\n                                             raise_exception=False,\n                                             default='DEBUG').upper()),\n                    format='%(asctime)s\\t%(process)d\\t%(levelname)s\\t%(message)s')\n\nGRACEFUL_STOP = threading.Event()\n\n\ndef reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False, scheme=None):\n    \"\"\"\n    Main loop to select and delete files.\n\n    :param rses: List of RSEs the reaper should work against. If empty, it considers all RSEs.\n    :param worker_number: The worker number.\n    :param total_workers:  The total number of workers.\n    :param chunk_size: the size of chunk for deletion.\n    :param once: If True, only runs one iteration of the main loop.\n    :param scheme: Force the reaper to use a particular protocol, e.g., mock.\n    \"\"\"\n    logging.info('Starting Dark Reaper %s-%s: Will work on RSEs: %s', worker_number, total_workers, ', '.join(rses))\n\n    pid = os.getpid()\n    thread = threading.current_thread()\n    hostname = socket.gethostname()\n    executable = ' '.join(sys.argv)\n    hash_executable = hashlib.sha256(sys.argv[0] + ''.join(rses)).hexdigest()\n    sanity_check(executable=None, hostname=hostname)\n\n    while not GRACEFUL_STOP.is_set():\n        try:\n            # heartbeat\n            heartbeat = live(executable=executable, hostname=hostname, pid=pid, thread=thread, hash_executable=hash_executable)\n            logging.info('Dark Reaper({0[worker_number]}/{0[total_workers]}): Live gives {0[heartbeat]}'.format(locals()))\n            nothing_to_do = True\n\n            random.shuffle(rses)\n            for rse_id in rses:\n                replicas = list_quarantined_replicas(rse_id=rse_id,\n                                                     limit=chunk_size, worker_number=worker_number,\n                                                     total_workers=total_workers)\n\n                rse_info = rsemgr.get_rse_info(rse_id=rse_id)\n                rse = rse_info['rse']\n                prot = rsemgr.create_protocol(rse_info, 'delete', scheme=scheme)\n                deleted_replicas = []\n                try:\n                    prot.connect()\n                    for replica in replicas:\n                        nothing_to_do = False\n                        scope = ''\n                        if replica['scope']:\n                            scope = replica['scope'].external\n                        try:\n                            pfn = str(rsemgr.lfns2pfns(rse_settings=rse_info,\n                                                       lfns=[{'scope': scope, 'name': replica['name'], 'path': replica['path']}],\n                                                       operation='delete', scheme=scheme).values()[0])\n                            logging.info('Dark Reaper %s-%s: Deletion ATTEMPT of %s:%s as %s on %s', worker_number, total_workers, scope, replica['name'], pfn, rse)\n                            start = time.time()\n                            prot.delete(pfn)\n                            duration = time.time() - start\n                            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)\n                            payload = {'scope': scope,\n                                       'name': replica['name'],\n                                       'rse': rse,\n                                       'rse_id': rse_id,\n                                       'file-size': replica.get('bytes') or 0,\n                                       'bytes': replica.get('bytes') or 0,\n                                       'url': pfn,\n                                       'duration': duration,\n                                       'protocol': prot.attributes['scheme']}\n                            if replica['scope'].vo != 'def':\n                                payload['vo'] = replica['scope'].vo\n                            add_message('deletion-done', payload)\n                            deleted_replicas.append(replica)\n                        except SourceNotFound:\n                            err_msg = 'Dark Reaper %s-%s: Deletion NOTFOUND of %s:%s as %s on %s' % (worker_number, total_workers, scope, replica['name'], pfn, rse)\n                            logging.warning(err_msg)\n                            deleted_replicas.append(replica)\n                        except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:\n                            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))\n                            logging.warning(err_msg)\n                            payload = {'scope': scope,\n                                       'name': replica['name'],\n                                       'rse': rse,\n                                       'rse_id': rse_id,\n                                       'file-size': replica['bytes'] or 0,\n                                       'bytes': replica['bytes'] or 0,\n                                       'url': pfn,\n                                       'reason': str(error),\n                                       'protocol': prot.attributes['scheme']}\n                            if replica['scope'].vo != 'def':\n                                payload['vo'] = replica['scope'].vo\n                            add_message('deletion-failed', payload)\n\n                        except Exception:\n                            logging.critical(traceback.format_exc())\n                finally:\n                    prot.close()\n\n                delete_quarantined_replicas(rse_id=rse_id, replicas=deleted_replicas)\n\n                if once:\n                    break\n\n            if once:\n                break\n\n            if nothing_to_do:\n                logging.info('Dark Reaper %s-%s: Nothing to do. I will sleep for 60s', worker_number, total_workers)\n                time.sleep(60)\n\n        except DatabaseException as error:\n            logging.warning('Reaper:  %s', str(error))\n        except Exception:\n            logging.critical(traceback.format_exc())\n\n    die(executable=executable, hostname=hostname, pid=pid, thread=thread, hash_executable=hash_executable)\n    logging.info('Graceful stop requested')\n    logging.info('Graceful stop done')\n    return\n\n\ndef stop(signum=None, frame=None):\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n    GRACEFUL_STOP.set()\n\n\ndef run(total_workers=1, chunk_size=100, once=False, rses=[], scheme=None,\n        exclude_rses=None, include_rses=None, vos=None, delay_seconds=0):\n    \"\"\"\n    Starts up the reaper threads.\n\n    :param total_workers: The total number of workers.\n    :param chunk_size: the size of chunk for deletion.\n    :param once: If True, only runs one iteration of the main loop.\n    :param rses: List of RSEs the reaper should work against. If empty, it considers all RSEs (Single-VO only).\n    :param scheme: Force the reaper to use a particular protocol/scheme, e.g., mock.\n    :param exclude_rses: RSE expression to exclude RSEs from the Reaper.\n    :param include_rses: RSE expression to include RSEs.\n    :param vos: VOs on which to look for RSEs. Only used in multi-VO mode.\n                If None, we either use all VOs if run from \"def\", or the current VO otherwise.\n    \"\"\"\n    if rucio.db.sqla.util.is_old_db():\n        raise exception.DatabaseException('Database was not updated, daemon won\\'t start')\n\n    logging.info('main: starting processes')\n\n    multi_vo = config_get_bool('common', 'multi_vo', raise_exception=False, default=False)\n    if not multi_vo:\n        if vos:\n            logging.warning('Ignoring argument vos, this is only applicable in a multi-VO setup.')\n        vos = ['def']\n    else:\n        if vos:\n            invalid = set(vos) - set([v['vo'] for v in list_vos()])\n            if invalid:\n                msg = 'VO{} {} cannot be found'.format('s' if len(invalid) > 1 else '', ', '.join([repr(v) for v in invalid]))\n                raise VONotFound(msg)\n        else:\n            vos = [v['vo'] for v in list_vos()]\n        logging.info('Dark Reaper: This instance will work on VO%s: %s' % ('s' if len(vos) > 1 else '', ', '.join([v for v in vos])))\n\n    all_rses = []\n    for vo in vos:\n        all_rses.extend(list_rses(filters={'vo': vo}))\n\n    if rses:\n        invalid = set(rses) - set([rse['rse'] for rse in all_rses])\n        if invalid:\n            msg = 'RSE{} {} cannot be found'.format('s' if len(invalid) > 1 else '',\n                                                    ', '.join([repr(rse) for rse in invalid]))\n            raise RSENotFound(msg)\n        rses = [rse for rse in all_rses if rse['rse'] in rses]\n    else:\n        rses = all_rses\n\n    if exclude_rses:\n        excluded_rses = parse_expression(exclude_rses)\n        rses = [rse for rse in rses if rse not in excluded_rses]\n\n    if include_rses:\n        included_rses = parse_expression(include_rses)\n        rses = [rse for rse in rses if rse in included_rses]\n\n    if not rses:\n        logging.error('Dark Reaper: No RSEs found. Exiting.')\n        return\n\n    threads = []\n    for worker in range(total_workers):\n        kwargs = {'worker_number': worker,\n                  'total_workers': total_workers,\n                  'rses': rses,\n                  'once': once,\n                  'chunk_size': chunk_size,\n                  'scheme': scheme}\n        threads.append(threading.Thread(target=reaper, kwargs=kwargs, name='Worker: %s, Total_Workers: %s' % (worker, total_workers)))\n    [t.start() for t in threads]\n    while threads[0].is_alive():\n        [t.join(timeout=3.14) for t in threads]\n", "path": "lib/rucio/daemons/reaper/dark_reaper.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2016-2021 CERN\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n# Authors:\n# - Vincent Garonne <vincent.garonne@cern.ch>, 2016-2018\n# - Martin Barisits <martin.barisits@cern.ch>, 2016\n# - Thomas Beermann <thomas.beermann@cern.ch>, 2016-2019\n# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019\n# - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019\n# - Cedric Serfon <cedric.serfon@cern.ch>, 2020\n# - Brandon White <bjwhite@fnal.gov>, 2019\n# - Patrick Austin <patrick.austin@stfc.ac.uk>, 2020\n# - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020\n# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020-2021\n\n'''\nDark Reaper is a daemon to manage quarantined file deletion.\n'''\n\nimport hashlib\nimport logging\nimport os\nimport random\nimport socket\nimport sys\nimport threading\nimport time\nimport traceback\n\nimport rucio.db.sqla.util\nfrom rucio.common import exception\nfrom rucio.common.config import config_get, config_get_bool\nfrom rucio.common.exception import (SourceNotFound, DatabaseException, ServiceUnavailable,\n                                    RSEAccessDenied, ResourceTemporaryUnavailable,\n                                    RSENotFound, VONotFound)\nfrom rucio.core.heartbeat import live, die, sanity_check\nfrom rucio.core.message import add_message\nfrom rucio.core.quarantined_replica import (list_quarantined_replicas,\n                                            delete_quarantined_replicas,\n                                            list_rses)\nfrom rucio.core.rse_expression_parser import parse_expression\nfrom rucio.core.vo import list_vos\nfrom rucio.rse import rsemanager as rsemgr\n\nlogging.getLogger(\"requests\").setLevel(logging.CRITICAL)\n\nlogging.basicConfig(stream=sys.stdout,\n                    level=getattr(logging,\n                                  config_get('common', 'loglevel',\n                                             raise_exception=False,\n                                             default='DEBUG').upper()),\n                    format='%(asctime)s\\t%(process)d\\t%(levelname)s\\t%(message)s')\n\nGRACEFUL_STOP = threading.Event()\n\n\ndef reaper(rses=[], worker_number=0, total_workers=1, chunk_size=100, once=False, scheme=None):\n    \"\"\"\n    Main loop to select and delete files.\n\n    :param rses: List of RSEs the reaper should work against. If empty, it considers all RSEs.\n    :param worker_number: The worker number.\n    :param total_workers:  The total number of workers.\n    :param chunk_size: the size of chunk for deletion.\n    :param once: If True, only runs one iteration of the main loop.\n    :param scheme: Force the reaper to use a particular protocol, e.g., mock.\n    \"\"\"\n    logging.info('Starting Dark Reaper %s-%s: Will work on RSEs: %s', worker_number, total_workers, ', '.join(rses))\n\n    pid = os.getpid()\n    thread = threading.current_thread()\n    hostname = socket.gethostname()\n    executable = ' '.join(sys.argv)\n    hash_executable = hashlib.sha256((sys.argv[0] + ''.join(rses)).encode()).hexdigest()\n    sanity_check(executable=None, hostname=hostname)\n\n    while not GRACEFUL_STOP.is_set():\n        try:\n            # heartbeat\n            heartbeat = live(executable=executable, hostname=hostname, pid=pid, thread=thread, hash_executable=hash_executable)\n            logging.info('Dark Reaper({0[worker_number]}/{0[total_workers]}): Live gives {0[heartbeat]}'.format(locals()))\n            nothing_to_do = True\n\n            random.shuffle(rses)\n            for rse_id in rses:\n                replicas = list_quarantined_replicas(rse_id=rse_id,\n                                                     limit=chunk_size, worker_number=worker_number,\n                                                     total_workers=total_workers)\n\n                rse_info = rsemgr.get_rse_info(rse_id=rse_id)\n                rse = rse_info['rse']\n                prot = rsemgr.create_protocol(rse_info, 'delete', scheme=scheme)\n                deleted_replicas = []\n                try:\n                    prot.connect()\n                    for replica in replicas:\n                        nothing_to_do = False\n                        scope = ''\n                        if replica['scope']:\n                            scope = replica['scope'].external\n                        try:\n                            pfn = str(list(rsemgr.lfns2pfns(rse_settings=rse_info,\n                                                            lfns=[{'scope': scope,\n                                                                   'name': replica['name'],\n                                                                   'path': replica['path']}],\n                                                            operation='delete',\n                                                            scheme=scheme).values())[0])\n                            logging.info('Dark Reaper %s-%s: Deletion ATTEMPT of %s:%s as %s on %s', worker_number, total_workers, scope, replica['name'], pfn, rse)\n                            start = time.time()\n                            prot.delete(pfn)\n                            duration = time.time() - start\n                            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)\n                            payload = {'scope': scope,\n                                       'name': replica['name'],\n                                       'rse': rse,\n                                       'rse_id': rse_id,\n                                       'file-size': replica.get('bytes') or 0,\n                                       'bytes': replica.get('bytes') or 0,\n                                       'url': pfn,\n                                       'duration': duration,\n                                       'protocol': prot.attributes['scheme']}\n                            if replica['scope'].vo != 'def':\n                                payload['vo'] = replica['scope'].vo\n                            add_message('deletion-done', payload)\n                            deleted_replicas.append(replica)\n                        except SourceNotFound:\n                            err_msg = 'Dark Reaper %s-%s: Deletion NOTFOUND of %s:%s as %s on %s' % (worker_number, total_workers, scope, replica['name'], pfn, rse)\n                            logging.warning(err_msg)\n                            deleted_replicas.append(replica)\n                        except (ServiceUnavailable, RSEAccessDenied, ResourceTemporaryUnavailable) as error:\n                            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))\n                            logging.warning(err_msg)\n                            payload = {'scope': scope,\n                                       'name': replica['name'],\n                                       'rse': rse,\n                                       'rse_id': rse_id,\n                                       'file-size': replica['bytes'] or 0,\n                                       'bytes': replica['bytes'] or 0,\n                                       'url': pfn,\n                                       'reason': str(error),\n                                       'protocol': prot.attributes['scheme']}\n                            if replica['scope'].vo != 'def':\n                                payload['vo'] = replica['scope'].vo\n                            add_message('deletion-failed', payload)\n\n                        except Exception:\n                            logging.critical(traceback.format_exc())\n                finally:\n                    prot.close()\n\n                delete_quarantined_replicas(rse_id=rse_id, replicas=deleted_replicas)\n\n                if once:\n                    break\n\n            if once:\n                break\n\n            if nothing_to_do:\n                logging.info('Dark Reaper %s-%s: Nothing to do. I will sleep for 60s', worker_number, total_workers)\n                time.sleep(60)\n\n        except DatabaseException as error:\n            logging.warning('Reaper:  %s', str(error))\n        except Exception:\n            logging.critical(traceback.format_exc())\n\n    die(executable=executable, hostname=hostname, pid=pid, thread=thread, hash_executable=hash_executable)\n    logging.info('Graceful stop requested')\n    logging.info('Graceful stop done')\n    return\n\n\ndef stop(signum=None, frame=None):\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n    GRACEFUL_STOP.set()\n\n\ndef run(total_workers=1, chunk_size=100, once=False, rses=[], scheme=None,\n        exclude_rses=None, include_rses=None, vos=None, delay_seconds=0):\n    \"\"\"\n    Starts up the reaper threads.\n\n    :param total_workers: The total number of workers.\n    :param chunk_size: the size of chunk for deletion.\n    :param once: If True, only runs one iteration of the main loop.\n    :param rses: List of RSEs the reaper should work against. If empty, it considers all RSEs (Single-VO only).\n    :param scheme: Force the reaper to use a particular protocol/scheme, e.g., mock.\n    :param exclude_rses: RSE expression to exclude RSEs from the Reaper.\n    :param include_rses: RSE expression to include RSEs.\n    :param vos: VOs on which to look for RSEs. Only used in multi-VO mode.\n                If None, we either use all VOs if run from \"def\", or the current VO otherwise.\n    \"\"\"\n    if rucio.db.sqla.util.is_old_db():\n        raise exception.DatabaseException('Database was not updated, daemon won\\'t start')\n\n    logging.info('main: starting processes')\n\n    multi_vo = config_get_bool('common', 'multi_vo', raise_exception=False, default=False)\n    if not multi_vo:\n        if vos:\n            logging.warning('Ignoring argument vos, this is only applicable in a multi-VO setup.')\n        vos = ['def']\n    else:\n        if vos:\n            invalid = set(vos) - set([v['vo'] for v in list_vos()])\n            if invalid:\n                msg = 'VO{} {} cannot be found'.format('s' if len(invalid) > 1 else '', ', '.join([repr(v) for v in invalid]))\n                raise VONotFound(msg)\n        else:\n            vos = [v['vo'] for v in list_vos()]\n        logging.info('Dark Reaper: This instance will work on VO%s: %s' % ('s' if len(vos) > 1 else '', ', '.join([v for v in vos])))\n\n    all_rses = []\n    for vo in vos:\n        all_rses.extend(list_rses(filters={'vo': vo}))\n\n    if rses:\n        invalid = set(rses) - set([rse['rse'] for rse in all_rses])\n        if invalid:\n            msg = 'RSE{} {} cannot be found'.format('s' if len(invalid) > 1 else '',\n                                                    ', '.join([repr(rse) for rse in invalid]))\n            raise RSENotFound(msg)\n        rses = [rse for rse in all_rses if rse['rse'] in rses]\n    else:\n        rses = all_rses\n\n    if exclude_rses:\n        excluded_rses = parse_expression(exclude_rses)\n        rses = [rse for rse in rses if rse not in excluded_rses]\n\n    if include_rses:\n        included_rses = parse_expression(include_rses)\n        rses = [rse for rse in rses if rse in included_rses]\n\n    if not rses:\n        logging.error('Dark Reaper: No RSEs found. Exiting.')\n        return\n\n    threads = []\n    for worker in range(total_workers):\n        kwargs = {'worker_number': worker,\n                  'total_workers': total_workers,\n                  'rses': rses,\n                  'once': once,\n                  'chunk_size': chunk_size,\n                  'scheme': scheme}\n        threads.append(threading.Thread(target=reaper, kwargs=kwargs, name='Worker: %s, Total_Workers: %s' % (worker, total_workers)))\n    [t.start() for t in threads]\n    while threads[0].is_alive():\n        [t.join(timeout=3.14) for t in threads]\n", "path": "lib/rucio/daemons/reaper/dark_reaper.py"}]}