From ea074ff176883062ef8336af4f7d486841aff642 Mon Sep 17 00:00:00 2001 From: voetberg Date: Wed, 18 Mar 2026 10:42:40 -0500 Subject: [PATCH 1/5] Common: Update to python3 Issue: probes#127 --- common/check_messages_to_submit | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/common/check_messages_to_submit b/common/check_messages_to_submit index 4400531..42b32a8 100755 --- a/common/check_messages_to_submit +++ b/common/check_messages_to_submit @@ -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"); @@ -12,7 +12,6 @@ """ Probe to check the queues of messages to submit by Hermes to the broker """ -from __future__ import print_function import sys From a672612e3f2792b11e906d2f115770dc38638524 Mon Sep 17 00:00:00 2001 From: voetberg Date: Wed, 18 Mar 2026 10:43:19 -0500 Subject: [PATCH 2/5] Common: Correct copyright header Issue: probes#127 --- common/check_messages_to_submit | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/common/check_messages_to_submit b/common/check_messages_to_submit index 42b32a8..1df059a 100755 --- a/common/check_messages_to_submit +++ b/common/check_messages_to_submit @@ -1,13 +1,17 @@ #!/usr/bin/env python3 -# Copyright European Organization for Nuclear Research (CERN) 2013 +# 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: -# - Mario Lassnig, , 2013-2014 -# - Thomas Beermann, , 2019 +# 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 queues of messages to submit by Hermes to the broker From e977df55fb83957d5cf9feeb76ae248e657518d2 Mon Sep 17 00:00:00 2001 From: voetberg Date: Wed, 18 Mar 2026 10:46:53 -0500 Subject: [PATCH 3/5] Common: Replace query with sqla2.0 format Issue: probes#127 --- common/check_messages_to_submit | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/common/check_messages_to_submit b/common/check_messages_to_submit index 1df059a..770d437 100755 --- a/common/check_messages_to_submit +++ b/common/check_messages_to_submit @@ -20,20 +20,17 @@ Probe to check the queues of messages to submit by Hermes to the broker import sys from prometheus_client import CollectorRegistry, Gauge, push_to_gateway +from sqlalchemy import func, select + from rucio.common.config import config_get -from rucio.db.sqla.session import BASE, get_session +from rucio.db.sqla import models +from rucio.db.sqla.session import get_session from utils.common import probe_metrics # Exit statuses OK, WARNING, CRITICAL, UNKNOWN = 0, 1, 2, 3 -if BASE.metadata.schema: - schema = BASE.metadata.schema + '.' -else: - schema = '' - -queue_sql = """SELECT COUNT(*) FROM {schema}messages""".format(schema=schema) PROM_SERVERS = config_get('monitor', 'prometheus_servers', raise_exception=False, default='') if PROM_SERVERS != '': @@ -43,7 +40,12 @@ if __name__ == "__main__": try: registry = CollectorRegistry() session = get_session() - result = session.execute(queue_sql).fetchall() + stmt = select( + func.count() + ).select_from( + models.Message + ) + result = session.execute(stmt).fetchall() print('queues.messages %s' % result[0][0]) probe_metrics.gauge(name='queues.messages').set(result[0][0]) Gauge('hermes_queues_messages', '', registry=registry).set(result[0][0]) From feb88daa914f1076987335e38b333aa547069df5 Mon Sep 17 00:00:00 2001 From: voetberg Date: Wed, 18 Mar 2026 10:48:14 -0500 Subject: [PATCH 4/5] Common: Replace 'fetch_all()' with 'scalar_one', remove extra indexing Issue: probes#127 --- common/check_messages_to_submit | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/common/check_messages_to_submit b/common/check_messages_to_submit index 770d437..cf20de7 100755 --- a/common/check_messages_to_submit +++ b/common/check_messages_to_submit @@ -45,10 +45,10 @@ if __name__ == "__main__": ).select_from( models.Message ) - result = session.execute(stmt).fetchall() - print('queues.messages %s' % result[0][0]) - probe_metrics.gauge(name='queues.messages').set(result[0][0]) - Gauge('hermes_queues_messages', '', registry=registry).set(result[0][0]) + result = session.execute(stmt).scalar_one() + print('queues.messages %s' % result) + probe_metrics.gauge(name='queues.messages').set(result) + Gauge('hermes_queues_messages', '', registry=registry).set(result) if len(PROM_SERVERS): for server in PROM_SERVERS: @@ -57,9 +57,9 @@ if __name__ == "__main__": except: continue - if result[0][0] > 100000: + if result > 100000: sys.exit(WARNING) - elif result[0][0] > 1000000: + elif result > 1000000: sys.exit(CRITICAL) except Exception as e: From 8678842286b9baba0fc518cca531bac704e09a14 Mon Sep 17 00:00:00 2001 From: voetberg Date: Wed, 18 Mar 2026 10:57:54 -0500 Subject: [PATCH 5/5] Common: Replace old prometheus approach with PromoetheusPusher Issue: probes#127 --- common/check_messages_to_submit | 35 +++++++++++---------------------- 1 file changed, 11 insertions(+), 24 deletions(-) diff --git a/common/check_messages_to_submit b/common/check_messages_to_submit index cf20de7..675bd2a 100755 --- a/common/check_messages_to_submit +++ b/common/check_messages_to_submit @@ -19,43 +19,30 @@ Probe to check the queues of messages to submit by Hermes to the broker import sys -from prometheus_client import CollectorRegistry, Gauge, push_to_gateway from sqlalchemy import func, select -from rucio.common.config import config_get from rucio.db.sqla import models from rucio.db.sqla.session import get_session -from utils.common import probe_metrics +from utils.common import PrometheusPusher # Exit statuses OK, WARNING, CRITICAL, UNKNOWN = 0, 1, 2, 3 -PROM_SERVERS = config_get('monitor', 'prometheus_servers', raise_exception=False, default='') -if PROM_SERVERS != '': - PROM_SERVERS = PROM_SERVERS.split(',') - if __name__ == "__main__": try: - registry = CollectorRegistry() - session = get_session() - stmt = select( - func.count() - ).select_from( - models.Message - ) - result = session.execute(stmt).scalar_one() - print('queues.messages %s' % result) - probe_metrics.gauge(name='queues.messages').set(result) - Gauge('hermes_queues_messages', '', registry=registry).set(result) - if len(PROM_SERVERS): - for server in PROM_SERVERS: - try: - push_to_gateway(server.strip(), job='check_messages_to_submit', registry=registry) - except: - continue + session = get_session() + with PrometheusPusher(job_name='hermes_queues_messages') as manager: + stmt = select( + func.count() + ).select_from( + models.Message + ) + result = session.execute(stmt).scalar_one() + print('queues.messages %s' % result) + manager.gauge(name='queues.messages').set(result) if result > 100000: sys.exit(WARNING)