From c456b4250bd871a5f0ee6e4bdc2c229ccdcf5a47 Mon Sep 17 00:00:00 2001
From: Dimitrios Christidis <dimitrios.christidis@cern.ch>
Date: Mon, 5 Aug 2024 14:15:02 +0200
Subject: [PATCH] Common: Rewrite check_obsolete_replicas

This is a complete rewrite that still maintains the exact same
functionality. Instead of one large ATLAS-specific PL/SQL script, this
implementation:

 1. Uses the gateway layer to list the RSEs and to update the RSE usage
    counters
 2. Works on each RSE separately
 3. Uses SQLAlchemy 2.0 syntax
 4. Replaces the deprecated function-based `REPLICAS_TOMBSTONE_IDX`
    index
---
 common/check_obsolete_replicas | 109 ++++++++++++---------------------
 1 file changed, 39 insertions(+), 70 deletions(-)

diff --git a/common/check_obsolete_replicas b/common/check_obsolete_replicas
index efd83943..35cf6d20 100755
--- a/common/check_obsolete_replicas
+++ b/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)