Common: Replace string formatter with f-strings, format context message uniformly #129

            Changes to metric names include:
            * undertaker_expired_dids -> expired_dids.total
            * fts3.{hostname}.submitted -> fts_backlog.submitted.{hostname}
            * hermes_queues_messages.queues.messages -> messages_to_submit.queues.messages
            * transmogrifier_new_dids -> new_dids
            * judge_stuck_rules_without_missing_source_replica -> stuck_rules.{source_status} (source_status = [without_missing_source_replica, with_missing_source_replica])
            * check_transfer_queues_status -> transfer_queues_status
            * judge.waiting_dids -> unevaluated_dids
            * reaper.unlocked_replicas -> unlocked_replicas.{replica_status} (replica_status = [expired, unlocked])
            * judge.updated_dids -> updated_dids

Author: voetberg

common/check_expired_dids

@@ -41,7 +41,8 @@ if __name__ == "__main__":
             # Possible check against a threshold. If result > max_value then sys.exit(CRITICAL)
 
             manager.gauge('expired_dids.total',
-                          documentation="All expired dids").set(result)
+                          documentation="All expired dids"
+                          ).set(result)
 
     except:
         print(traceback.format_exc())

common/check_fts_backlog

@@ -78,11 +78,11 @@ if __name__ == "__main__":
 
         errmsg = ''
         for ftshost in FTSHOSTS.split(','):
-            print("=== %s ===" % ftshost)
+            print(f"=== {ftshost} ===")
             parsed_url = urlparse(ftshost)
             scheme, hostname, port = parsed_url.scheme, parsed_url.hostname, parsed_url.port
             retvalue = CRITICAL
-            url = '%s/fts3/ftsmon/overview?dest_se=&source_se=&time_window=1&vo=%s' % (ftshost, VO)
+            url = f'{ftshost}/fts3/ftsmon/overview?dest_se=&source_se=&time_window=1&vo={VO}'
             busy_channels = []
             busylimit = 5000
             for attempt in range(0, 5):
@@ -108,7 +108,7 @@ if __name__ == "__main__":
                             pass
 
                         if CHECK_BUSY and 'submitted' in channel and channel['submitted'] >= busylimit:
-                            url_activities = '%s/fts3/ftsmon/config/activities/%s?source_se=%s&dest_se=%s' % (ftshost, VO, src, dst)
+                            url_activities = f'{ftshost}/fts3/ftsmon/config/activities/{VO}?source_se={src}&dest_se={dst}'
                             activities = {}
                             try:
                                 s = requests.get(url_activities, verify=False, cert=(PROXY, PROXY))
@@ -120,36 +120,33 @@ if __name__ == "__main__":
                                                   'activities': activities})
                     summary = res['summary']
                     hostname = hostname.replace('.', '_')
-                    # If printing these indiv is important, why not monitor them seperately? 
-                    print('%s : Submitted : %s' % (hostname, summary['submitted']))
-                    print('%s : Active : %s' % (hostname, summary['active']))
-                    print('%s : Staging : %s' % (hostname, summary['staging']))
-                    print('%s : Started : %s' % (hostname, summary['started']))
+
+                    for state in ['submitted', 'active', 'staging', 'started']: 
+                        print(f'{hostname} : {state.capitalize()} : {summary[state]}')
+
 
                     if busy_channels != []:
-                        print('Busy channels (>%s submitted):' % busylimit)
+                        print(f'Busy channels (>{busylimit} submitted):')
                         for bc in busy_channels:
-                            activities_str = ", ".join([("%s: %s" % (key, val)) for key, val in bc['activities'].items()])
-                            print('  %s to %s : %s submitted jobs (%s)' % (bc['src'], bc['dst'], bc['submitted'],
-                                                                           str(activities_str)))
+                            activities_str = ", ".join([(f"{key}: {val}") for key, val in bc['activities'].items()])
+                            print(f'{bc['src']} to {bc['dst']} : {bc['submitted']} submitted jobs {activities_str}')
 
                     # Add to metrics
                     backlog_count = summary['submitted'] + summary['active'] + summary['staging'] + summary['started']
-                    manager.gauge(
-                        "fts_backlog.submitted.{hostname}", 
-                        documentation="All submitted, active, staged, or stated in FTS queue").labels(hostname=hostname).set(backlog_count)
+                    manager.gauge("fts_backlog.submitted.{hostname}", 
+                        documentation="All submitted, active, staged, or stated in FTS queue"
+                        ).labels(hostname=hostname).set(backlog_count)
 
                     retvalue = OK
                     break
                 except Exception as error:
                     retvalue = CRITICAL
                     if result and result.status_code:
-                        errmsg = 'Error when trying to get info from %s : HTTP status code %s. [%s]' % (
-                            ftshost, str(result.status_code), str(error))
+                        errmsg = f'Error when trying to get info from {ftshost} : HTTP status code {result.status_code}. {error}'
                     else:
-                        errmsg = 'Error when trying to get info from %s. %s' % (ftshost, str(error))
+                        errmsg = f'Error when trying to get info from {ftshost}. {error}'
             if retvalue == CRITICAL:
-                print("All attempts failed. %s" % errmsg)
+                print(f"All attempts failed. {errmsg}")
             WORST_RETVALUE = max(retvalue, WORST_RETVALUE)
 
 
@@ -179,7 +176,6 @@ if __name__ == "__main__":
     except:
         sys.exit(WORST_RETVALUE)
 
-    # Does this not do the same thing as the query? Why the duplicate? 
     for source_rse, dest_rse in se_matrix:
         for source_rse_id in se_map[source_rse]:
             for dest_rse_id in se_map[dest_rse]:

common/check_messages_to_submit

@@ -41,7 +41,8 @@ if __name__ == "__main__":
 
             manager.gauge(
                 "messages_to_submit.queues.messages", 
-                documentation="Messages in queue, to submit").set(message_count)
+                documentation="Messages in queue, to submit"
+                ).set(message_count)
 
         if message_count > 100000:
             sys.exit(WARNING)

common/check_obsolete_replicas

@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 # Copyright European Organization for Nuclear Research (CERN) 2013
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -8,77 +8,101 @@
 # Authors:
 # - Vincent Garonne, <vincent.garonne@cern.ch>, 2015
 # - Cedric Serfon, <cedric.serfon@cern.ch>, 2018
+# - Maggie Voetberg, <maggiev@fnal.gov>, 2024
 
 '''
 Probe to check the backlog of obsolete replicas.
 '''
 
 import sys
+import traceback
+from sqlalchemy.sql import text
+from rucio.db.sqla.session import BASE, get_session
+from utils.common import PrometheusPusher
 
-from rucio.db.sqla.session import get_session
+if BASE.metadata.schema:
+    schema = BASE.metadata.schema + '.'
+else:
+    schema = ''
 
 # Exit statuses
 OK, WARNING, CRITICAL, UNKNOWN = 0, 1, 2, 3
 
 
 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)
+        session = get_session()
+        with PrometheusPusher() as manager:
+            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
+                                {schema}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
+                                {schema}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;
+            LOOP
+                    MERGE INTO {schema}RSE_USAGE
+                    USING DUAL
+                    ON ({schema}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);
+                    MERGE INTO {schema}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
+                    COMMIT;
+            END LOOP;
+    END;'''.format(schema=schema)
+            
+            for result in session.execute(text(query)):
+                print(result)
+
+                rse_id = result[0]
+                bytes_sum = result[2]
+                files_count = result[3]
+
+                manager.gauge(name="obsolete_replicas_files.{rse}",
+                              documentation="Probe to check the backlog of obsolete replicas.").labels(rse=rse_id).set(files_count)
+                
+                manager.gauge(name="obsolete_replicas_bytes.{rse}",
+                                documentation="Probe to check the backlog of obsolete replicas.").labels().set(bytes_sum)
+
+
+    except:
+        print(traceback.format_exc())
         sys.exit(UNKNOWN)
+    finally:
+        session.remove()
     sys.exit(OK)

common/check_stuck_rules

@@ -44,9 +44,9 @@ if __name__ == "__main__":
         with PrometheusPusher() as manager:
             for source_status, query in queries.items(): 
                 result = session.execute(sql_text(query)).fetchone()[0]
-                manager.gauge(
-                    "stuck_rules.{source_status}", 
-                    documentation="Backlog of stuck rules").labels(source_status=source_status).set(result)
+                manager.gauge("stuck_rules.{source_status}", 
+                    documentation="Backlog of stuck rules"
+                    ).labels(source_status=source_status).set(result)
 
     except:
         print(traceback.format_exc())

common/check_transfer_queues_status

@@ -69,7 +69,9 @@ if __name__ == "__main__":
                 activity = items[3]
                 external_host = items[4]
 
-                manager.gauge("transfer_queues_status.{activity}.{state}.{external_host}").labels(activity=activity, state=state, external_host=external_host).set(count)
+                manager.gauge(
+                    "transfer_queues_status.{activity}.{state}.{external_host}"
+                    ).labels(activity=activity, state=state, external_host=external_host).set(count)
 
     except Exception as e:
         print(f"Error: {e}")

common/check_unlocked_replicas

@@ -41,9 +41,9 @@ if __name__ == "__main__":
         }
 
         with PrometheusPusher() as manager:
-            for did_status, query in queries.items(): 
+            for replica_status, query in queries.items(): 
                 result = session.execute(sql_text(query)).fetchone()[0]
-                manager.gauge("unlocked_dids.{did_status}").labels(did_status=did_status).set(result)
+                manager.gauge("unlocked_replicas.{replica_status}").labels(did_status=replica_status).set(result)
 
     except:
         sys.exit(UNKNOWN)