Common: Rewrite check_obsolete_replicas

common/check_obsolete_replicas

@@ -1,84 +1,53 @@
-#!/usr/bin/env python
-# Copyright European Organization for Nuclear Research (CERN) 2013
+#!/usr/bin/env python3
+# Copyright European Organization for Nuclear Research (CERN) since 2012
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
-# You may not use this file except in compliance with the License.
-# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
 #
-# Authors:
-# - Vincent Garonne, <vincent.garonne@cern.ch>, 2015
-# - Cedric Serfon, <cedric.serfon@cern.ch>, 2018
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
 
-'''
-Probe to check the backlog of obsolete replicas.
-'''
+"""
+Calculate the amount of obsolete replicas and update the RSE usage counters.
+"""
 
 import sys
+import traceback
+
+from sqlalchemy import func, select
 
+from rucio.db.sqla import models
+from rucio.db.sqla.constants import OBSOLETE
 from rucio.db.sqla.session import get_session
+from rucio.gateway.rse import list_rses, set_rse_usage
 
-# Exit statuses
 OK, WARNING, CRITICAL, UNKNOWN = 0, 1, 2, 3
 
-
-if __name__ == "__main__":
+if __name__ == '__main__':
     try:
-        SESSION = get_session()
-        QUERY = '''BEGIN
-        FOR u in (SELECT
-                      a.rse_id                        AS rse_id,
-                      NVL(b.files, 0)                 AS files,
-                      NVL(b.bytes, 0)                 AS bytes,
-                      SYS_EXTRACT_UTC(localtimestamp) AS updated_at
-                  FROM
-                      (
-                          SELECT
-                              id AS rse_id
-                          FROM
-                              atlas_rucio.rses
-                          WHERE
-                              deleted=0) a
-                  LEFT OUTER JOIN
-                      (
-                          SELECT
-                              /*+  INDEX_FFS(replicas REPLICAS_TOMBSTONE_IDX) */
-                              rse_id,
-                              COUNT(1)   AS files,
-                              SUM(bytes) AS bytes
-                          FROM
-                              atlas_rucio.replicas
-                          WHERE
-                              (
-                                  CASE
-                                      WHEN tombstone IS NOT NULL
-                                      THEN rse_id
-                                  END) IS NOT NULL
-                          AND tombstone=to_date('1-1-1970 00:00:00','MM-DD-YYYY HH24:Mi:SS')
-                          GROUP BY
-                              rse_id) b
-                  ON
-                      a.rse_id=b.rse_id)
-
-        LOOP
-                MERGE INTO atlas_rucio.RSE_USAGE
-                USING DUAL
-                ON (atlas_rucio.RSE_USAGE.rse_id = u.rse_id and source = 'obsolete')
-                WHEN NOT MATCHED THEN INSERT(rse_id, source, used, files, updated_at, created_at)
-                VALUES (u.rse_id, 'obsolete', u.bytes, u.files, u.updated_at, u.updated_at)
-                WHEN MATCHED THEN UPDATE SET used=u.bytes, files=u.files, updated_at=u.updated_at;
-
-                MERGE INTO ATLAS_RUCIO.RSE_USAGE_HISTORY H
-                USING DUAL
-                ON (h.rse_id = u.rse_id and h.source = 'obsolete' and h.updated_at = u.updated_at)
-                WHEN NOT MATCHED THEN INSERT(rse_id, source, used, files, updated_at, created_at)
-                VALUES (u.rse_id, 'obsolete', u.bytes, u.files, u.updated_at, u.updated_at);
-
-                COMMIT;
-        END LOOP;
-END;
-'''
-        SESSION.execute(QUERY)
-    except Exception as error:
-        print error
+        session = get_session()
+        for rse in list_rses(session=session):
+            query = select(
+                func.coalesce(func.sum(models.RSEFileAssociation.bytes), 0).label('bytes'),
+                func.count().label('files')
+            ).with_hint(
+                models.RSEFileAssociation, 'INDEX(replicas REPLICAS_RSE_ID_TOMBSTONE_IDX)', 'oracle'
+            ).where(
+                models.RSEFileAssociation.rse_id == rse['id'],
+                models.RSEFileAssociation.tombstone == OBSOLETE
+            )
+
+            bytes_, files = session.execute(query).one()
+            set_rse_usage(rse=rse['rse'], source='obsolete', used=bytes_,
+                          free=None, issuer='root', files=files,
+                          session=session)
+    except Exception:
+        print(traceback.format_exc())
         sys.exit(UNKNOWN)
     sys.exit(OK)