Skip to content

Commit

Permalink
Auth: get_updated_account_counters, migrate query to SQLA2.0 and retu…
Browse files Browse the repository at this point in the history
…rn as list

For more information, see: rucio#6497 (comment)
  • Loading branch information
rdimaio committed May 2, 2024
1 parent b316c12 commit b434f63
Show file tree
Hide file tree
Showing 2 changed files with 14 additions and 10 deletions.
10 changes: 7 additions & 3 deletions lib/rucio/core/account_counter.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,12 +90,16 @@ def get_updated_account_counters(total_workers, worker_number, *, session: "Sess
:param session: Database session in use.
:returns: List of rse_ids whose rse_counters need to be updated.
"""
query = session.query(models.UpdatedAccountCounter.account, models.UpdatedAccountCounter.rse_id).\
distinct(models.UpdatedAccountCounter.account, models.UpdatedAccountCounter.rse_id)

query = select(
models.UpdatedAccountCounter.account,
models.UpdatedAccountCounter.rse_id,
).distinct(
)

query = filter_thread_work(session=session, query=query, total_threads=total_workers, thread_id=worker_number, hash_variable='CONCAT(account, rse_id)')

return query.all()
return [row._asdict() for row in session.execute(query).all()]


@transactional_session
Expand Down
14 changes: 7 additions & 7 deletions lib/rucio/daemons/abacus/account.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,22 +54,22 @@ def run_once(heartbeat_handler, **_kwargs):
worker_number, total_workers, logger = heartbeat_handler.live()

start = time.time() # NOQA
account_rse_ids = get_updated_account_counters(total_workers=total_workers,
worker_number=worker_number)
logger(logging.DEBUG, 'Index query time %f size=%d' % (time.time() - start, len(account_rse_ids)))
updated_account_counters = get_updated_account_counters(total_workers=total_workers,
worker_number=worker_number)
logger(logging.DEBUG, 'Index query time %f size=%d' % (time.time() - start, len(updated_account_counters)))

# If the list is empty, sent the worker to sleep
if not account_rse_ids:
if not updated_account_counters:
logger(logging.INFO, 'did not get any work')
return

for account_rse_id in account_rse_ids:
for account_counter in updated_account_counters:
worker_number, total_workers, logger = heartbeat_handler.live()
if graceful_stop.is_set():
break
start_time = time.time()
update_account_counter(account=account_rse_id[0], rse_id=account_rse_id[1])
logger(logging.DEBUG, 'update of account-rse counter "%s-%s" took %f' % (account_rse_id[0], account_rse_id[1], time.time() - start_time))
update_account_counter(account=account_counter['account'], rse_id=account_counter['rse_id'])
logger(logging.DEBUG, 'update of account-rse counter "%s-%s" took %f' % (account_counter['account'], account_counter['rse_id'], time.time() - start_time))


def stop(signum: "Optional[int]" = None, frame: "Optional[FrameType]" = None) -> None:
Expand Down

0 comments on commit b434f63

Please sign in to comment.