opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

WIP pgcleanup improvements

Browse Source
This commit is contained in:
Nate Coraor
2018-10-15 14:59:15 -04:00
parent 2a422fa61f
commit 429ef96a10
1 changed files with 361 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/cleanup_datasets/pgcleanup.py
+361 -114
View File
@@ -13,8 +13,11 @@ import logging
import os
import shutil
import sys
from collections import namedtuple
from functools import partial
import psycopg2
from psycopg2.extras import NamedTupleCursor
from six import string_types
from sqlalchemy.engine.url import make_url
@@ -27,26 +30,55 @@ from galaxy.objectstore import build_object_store_from_config
from galaxy.util.bunch import Bunch
from galaxy.util.script import app_properties_from_args, populate_config_args
log = logging.getLogger()
log = logging.getLogger(__name__)
MetadataFile = namedtuple('MetadataFile', ['id', 'object_store_id'])
Dataset = namedtuple('Dataset', ['id', 'object_store_id'])
class MetadataFile(Bunch):
class RemovesMetadataFiles(object):
pass
class Dataset(Bunch):
class RemovesDatasets(object):
pass
class Action(object):
def handle_row(self, row):
pass
def post_action(self):
pass
class PurgeDatasets(RemovesDatasets, Action):
sql = """
"""
def __init__(self):
self.datasets = set()
def handle_row(self, row):
self.datasets.add(Dataset(row.purged_dataset_id, row.object_store_id))
def post_action(self):
pass
class Cleanup(object):
def __init__(self):
self.args = None
self.config = None
self.conn = None
self.action_names = []
self.logs = {}
self.disk_accounting_user_ids = []
self.disk_accounting_user_ids = set()
self.remove_metadata_files = set()
self.remove_datasets = set()
self.object_store = None
self.current_action = None
self.action_logger = None
self.__cache_action_names()
self.__parse_args()
@@ -63,7 +95,7 @@ class Cleanup(object):
def __parse_args(self):
parser = argparse.ArgumentParser()
populate_config_args(parser)
parser.add_argument('-d', '--debug', action='store_true', dest='debug', help='Enable debug logging', default=False)
parser.add_argument('-d', '--debug', action='store_true', dest='debug', help='Enable debug logging (SQL queries)', default=False)
parser.add_argument('--dry-run', action='store_true', dest='dry_run', help="Dry run (rollback all transactions)", default=False)
parser.add_argument('--force-retry', action='store_true', dest='force_retry', help="Retry file removals (on applicable actions)", default=False)
parser.add_argument('-o', '--older-than', type=int, dest='days', help='Only perform action(s) on objects that have not been updated since the specified number of days', default=14)
@@ -81,7 +113,7 @@ class Cleanup(object):
sys.exit(0)
def __setup_logging(self):
format = "%(funcName)s %(levelname)s %(asctime)s %(message)s"
format = "%(asctime)s %(levelname)-5s %(funcName)s: %(message)s"
if self.args.debug:
logging.basicConfig(level=logging.DEBUG, format=format)
else:
@@ -100,47 +132,80 @@ class Cleanup(object):
assert url.get_dialect().name == 'postgresql', 'This script can only be used with PostgreSQL.'
self.conn = psycopg2.connect(**args)
self.conn = psycopg2.connect(cursor_factory=NamedTupleCursor, **args)
def __load_object_store(self):
self.object_store = build_object_store_from_config(self.config)
def _open_logfile(self):
action_name = inspect.stack()[1][3]
logname = os.path.join(self.args.log_dir, action_name + '.log')
logname = os.path.join(self.args.log_dir, self.current_action + '.log')
if self.args.dry_run:
log.debug('--dry-run specified, logging changes to stdout instead of log file: %s' % logname)
self.logs[action_name] = sys.stdout
log.info('--dry-run specified, logging changes to stdout instead of log file: %s' % logname)
self._log_fh = sys.stdout
else:
log.debug('Opening log file: %s' % logname)
self.logs[action_name] = open(logname, 'a')
log.info('Opening log file: %s' % logname)
self._log_fh = open(logname, 'a')
message = '==== Log opened: %s ' % datetime.datetime.now().isoformat()
self.logs[action_name].write(message.ljust(72, '='))
self.logs[action_name].write('\n')
self._log_fh.write(message.ljust(72, '='))
self._log_fh.write('\n')
def _log(self, message, action_name=None):
if action_name is None:
action_name = inspect.stack()[1][3]
def _log(self, message):
self.action_log.info(message)
'''
if self._log_fh is None:
self._open_logfile()
if not message.endswith('\n'):
message += '\n'
self.logs[action_name].write(message)
self._log_fh.write(message)
'''
@property
def action_log(self):
if self.action_logger is None:
logf = os.path.join(self.args.log_dir, self.current_action + '.log')
if self.args.dry_run:
log.info('--dry-run specified, logging changes to stderr instead of log file: %s' % logf)
h = logging.StreamHandler()
else:
log.info('Opening log file: %s' % logf)
h = logging.FileHandler(logf)
h.setLevel(logging.DEBUG if self.args.debug else logging.INFO)
f = logging.Formatter('%(message)s')
h.setFormatter(f)
l = logging.getLogger(self.current_action)
l.addHandler(h)
l.propagate = False
m = ('==== Log opened: %s ' % datetime.datetime.now().isoformat()).ljust(72, '=')
l.info(m)
self.action_logger = l
return self.action_logger
def _close_log(self):
if self.action_logger is not None:
m = ('==== Log closed: %s ' % datetime.datetime.now().isoformat()).ljust(72, '=')
self.action_logger.info(m)
# TODO: close the handler?
self.action_logger = None
'''
def _close_logfile(self):
action_name = inspect.stack()[1][3]
if self._log_fh is None:
return
message = '==== Log closed: %s ' % datetime.datetime.now().isoformat()
self.logs[action_name].write(message.ljust(72, '='))
self.logs[action_name].write('\n')
self._log_fh.write(message.ljust(72, '='))
self._log_fh.write('\n')
if self.args.dry_run:
log.debug('--dry-run specified, changes were logged to stdout insted of log file')
log.info('--dry-run specified, changes were logged to stdout insted of log file')
else:
log.debug('Closing log file: %s' % self.logs[action_name].name)
self.logs[action_name].close()
log.info('Closing log file: %s' % self._log_fh.name)
self._log_fh.close()
del self.logs[action_name]
self._log_fh = None
'''
def _run(self):
ok = True
@@ -153,7 +218,9 @@ class Cleanup(object):
sys.exit(1)
for name in self.args.sequence:
log.info('Calling %s' % name)
self.current_action = name
self.__getattribute__(name)()
self._close_log()
log.info('Finished %s' % name)
def _create_event(self, message=None):
@@ -162,7 +229,7 @@ class Cleanup(object):
"""
if message is None:
message = inspect.stack()[2][3]
message = self.current_action
sql = """
INSERT INTO cleanup_event
@@ -198,23 +265,22 @@ class Cleanup(object):
return cur.fetchone()[0]
@property
def _update_time_sql(self):
# only update time if not doing force retry (otherwise a lot of things would have their update times reset that were actually purged a long time ago)
if self.args.update_time and not self.args.force_retry:
return ", update_time = NOW()"
return ""
@property
def _force_retry_sql(self):
def _force_retry_sql(self, force_retry_sql=None):
# a misnomer, the value of force_retry_sql is actually the SQL that should be used if *not* forcibly retrying
if not self.args.force_retry:
return " AND NOT purged"""
return force_retry_sql or " AND NOT purged"
return ""
def _format_sql(self, sql):
def _format_sql(self, sql, force_retry_sql=None):
return sql.format(
update_time_sql=self._update_time_sql,
force_retry_sql=self._force_retry_sql,
update_time_sql=self._update_time_sql(),
force_retry_sql=self._force_retry_sql(force_retry_sql=force_retry_sql),
)
def _add_default_args(self, args):
@@ -222,8 +288,8 @@ class Cleanup(object):
if 'days' not in args:
args['days'] = self.args.days
def _update(self, sql, args, add_event=True, event_message=None):
sql = self._format_sql(sql)
def _update(self, sql, args, add_event=True, event_message=None, force_retry_sql=None):
sql = self._format_sql(sql, force_retry_sql=force_retry_sql)
if add_event and isinstance(args, dict) and 'event_id' not in args:
args['event_id'] = self._create_event(message=event_message)
self._add_default_args(args)
@@ -242,6 +308,8 @@ class Cleanup(object):
log.info('Executing SQL')
cur.execute(sql, args)
log.info('Database status: %s' % cur.statusmessage)
log.info('Flushing transaction')
self._flush()
return cur
@@ -253,12 +321,60 @@ class Cleanup(object):
self.conn.commit()
log.info("All changes committed")
def _remove_metadata_file(self, id, object_store_id, action_name):
def _collect_row_results(self, row, results, causals, primary_key):
primary_key = row._fields[0]
primary = getattr(row, primary_key)
if primary not in results:
results[primary] = [set() for x in range(len(causals))]
rowgetter = partial(getattr, row)
for i, causal in enumerate(causals):
vals = tuple(map(rowgetter, causal))
if any(vals[1:]):
results[primary][i].add(vals)
def _collect_row_actions(self, row):
if hasattr(row, 'disk_accounting_user_id'):
self.disk_accounting_user_ids.add(row.disk_accounting_user_id)
object_store_id = getattr(row, 'object_store_id', None)
if hasattr(row, 'deleted_metadata_file_id'):
self.metadata_files.add((row.deleted_metadata_file_id, object_store_id))
def _log_results(self, results, causals, primary_key):
for primary in sorted(results.keys()):
self._log('%s: %s' % (primary_key, primary))
for causal, s in zip(causals, results[primary]):
for r in sorted(s):
secondaries = ', '.join(['%s: %s' % x for x in zip(causal[1:], r[1:])])
self._log('%s %s caused %s' % (causal[0], r[0], secondaries))
def _handle_results(self, cur, causals=None):
"""Log the results of an action where results are joined across many tables and may contain duplicates, and
handle any actions that should be taken on those results.
:param row: cursor object to iterate
:type row: :class:`psycopg2.cursor`
:param causals: names of columns that caused actions to be performed on other columns
:type causals: tuple of tuples of strs
The first column is considered the "primary" column upon which actions are being taken. If disk usage should be
recalculated for users, those user IDs should be in a column named ``disk_accounting_user_id``.
"""
causals = causals or ()
results = {}
primary_key = None
for row in cur:
if primary_key is None:
primary_key = row._fields[0]
self._collect_row_results(row, results, causals, primary_key)
self._collect_row_actions(row)
self._log_results(results, causals, primary_key)
def _remove_metadata_file(self, id, object_store_id):
metadata_file = MetadataFile(id=id, object_store_id=object_store_id)
try:
filename = self.object_store.get_filename(metadata_file, extra_dir='_metadata_files', extra_dir_at_root=True, alt_name="metadata_%d.dat" % id)
self._log('Removing from disk: %s' % filename, action_name)
self._log('Removing from disk: %s' % filename)
except (ObjectNotFound, AttributeError) as e:
log.error('Unable to get MetadataFile %s filename: %s' % (id, e))
return
@@ -267,7 +383,7 @@ class Cleanup(object):
try:
os.unlink(filename)
except Exception as e:
self._log('Removal of %s failed with error: %s' % (filename, e), action_name)
self._log('Removal of %s failed with error: %s' % (filename, e))
def _update_user_disk_usage(self):
"""
@@ -278,9 +394,9 @@ class Cleanup(object):
This could probably be done more efficiently.
"""
log.info('Recalculating disk usage for users whose HistoryDatasetAssociations were purged')
log.info('Recalculating disk usage for users whose data were purged')
for user_id in self.disk_accounting_user_ids:
for user_id in sorted(self.disk_accounting_user_ids):
# TODO: h.purged = false should be unnecessary once all hdas in purged histories are purged.
sql = """
@@ -303,20 +419,17 @@ class Cleanup(object):
args = {'user_id': user_id}
cur = self._update(sql, args, add_event=False)
self._flush()
for tup in cur:
# disk_usage might be None (e.g. user has purged all data)
log.debug('Updated disk usage for user id %i to %s bytes' % (user_id, tup[0]))
log.info('Updated disk usage for user id %i to %s bytes' % (user_id, tup[0]))
def _shutdown(self):
self.object_store.shutdown()
self.conn.close()
for handle in self.logs.values():
message = '==== Log closed at shutdown: %s ' % datetime.datetime.now().isoformat()
handle.write(message.ljust(72, '='))
handle.write('\n')
handle.close()
if self.action_log is not None:
m = ('==== Log closed at shutdown: %s ' % datetime.datetime.now().isoformat()).ljust(72, '=')
self.action_logger.info(m)
#
# actions
@@ -332,7 +445,6 @@ class Cleanup(object):
The old cleanup script does not mark HistoryDatasetAssociations as purged when deleted Histories are purged. This method can be used to rectify that situation.
"""
log.info('Marking purged all HistoryDatasetAssociations associated with purged Datasets')
# update_time is intentionally left unmodified.
sql = """
WITH purged_hda_ids
@@ -348,25 +460,204 @@ class Cleanup(object):
(create_time, cleanup_event_id, hda_id)
SELECT NOW(), %(event_id)s, id
FROM purged_hda_ids)
SELECT id
SELECT id AS purged_hda_id
FROM purged_hda_ids
ORDER BY id;
"""
cur = self._update(sql, {})
self._flush()
self._handle_results(cur)
self._open_logfile()
for tup in cur:
self._log('Marked HistoryDatasetAssociation purged: %s' % tup[0])
self._close_logfile()
def delete_inactive_users(self):
"""
Mark deleted all users that are older than the specified number of days.
Mark deleted (state = 'deleted') all Jobs whose user_ids are deleted in this step.
"""
log.info('Marking deleted all inactive GalaxyUsers older than %i days' % self.args.days)
sql = """
WITH deleted_user_ids
AS ( UPDATE galaxy_user
SET deleted = true{update_time_sql}
WHERE NOT active{force_retry_sql}
AND update_time < (NOW() - interval '%(days)s days')
RETURNING id),
deleted_job_ids
AS ( UPDATE job
SET state = 'deleted'{update_time_sql}
FROM deleted_user_ids
WHERE deleted_user_ids.id = job.user_id
AND job.state = 'new'
RETURNING job.user_id AS user_id,
job.id AS id),
user_events
AS (INSERT INTO cleanup_event_user_association
(create_time, cleanup_event_id, user_id)
SELECT NOW(), %(event_id)s, id
FROM deleted_user_ids)
SELECT deleted_user_ids.id AS deleted_user_id,
deleted_job_ids.id AS deleted_job_id
FROM deleted_user_ids
LEFT OUTER JOIN deleted_job_ids
ON deleted_job_ids.user_id = deleted_user_ids.id
ORDER BY deleted_user_ids.id
"""
force_retry_sql = " AND NOT deleted"
cur = self._update(sql, {}, force_retry_sql=force_retry_sql)
causals = (
('purged_user_id', 'purged_job_id'),
)
self._handle_results(cur, causals)
def purge_deleted_users(self):
"""
Mark purged all users that are older than the specified number of days.
Mark purged all Histories whose user_ids are purged in this step.
Mark purged all HistoryDatasetAssociations whose history_ids are purged in this step.
Delete all UserGroupAssociations whose user_ids are purged in this step.
Delete all UserRoleAssociations whose user_ids are purged in this step EXCEPT FOR THE PRIVATE ROLE.
Delete all UserAddresses whose user_ids are purged in this step.
"""
log.info('Marking purged all deleted GalaxyUsers older than %i days' % self.args.days)
sql = """
WITH purged_user_ids
AS ( UPDATE galaxy_user
SET purged = true{update_time_sql}
WHERE deleted{force_retry_sql}
AND update_time < (NOW() - interval '%(days)s days')
RETURNING id),
purged_history_ids
AS ( UPDATE history
SET purged = true{update_time_sql}
FROM purged_user_ids
WHERE purged_user_ids.id = history.user_id
AND NOT history.purged
RETURNING history.user_id AS user_id,
history.id AS id),
purged_hda_ids
AS ( UPDATE history_dataset_association
SET purged = true{update_time_sql}
FROM purged_history_ids
WHERE purged_history_ids.id = history_dataset_association.history_id
AND NOT history_dataset_association.purged
RETURNING history_dataset_association.history_id AS history_id,
history_dataset_association.id AS id),
deleted_metadata_file_ids
AS ( UPDATE metadata_file
SET deleted = true{update_time_sql}
FROM purged_hda_ids
WHERE purged_hda_ids.id = metadata_file.hda_id
RETURNING metadata_file.hda_id AS hda_id,
metadata_file.id AS id,
metadata_file.object_store_id AS object_store_id),
deleted_icda_ids
AS ( UPDATE implicitly_converted_dataset_association
SET deleted = true{update_time_sql}
FROM purged_hda_ids
WHERE purged_hda_ids.id = implicitly_converted_dataset_association.hda_parent_id
RETURNING implicitly_converted_dataset_association.hda_id AS hda_id,
implicitly_converted_dataset_association.hda_parent_id AS hda_parent_id,
implicitly_converted_dataset_association.id AS id),
deleted_icda_purged_child_hda_ids
AS ( UPDATE history_dataset_association
SET purged = true{update_time_sql}
FROM deleted_icda_ids
WHERE deleted_icda_ids.hda_id = history_dataset_association.id),
deleted_uga_ids
AS (DELETE FROM user_group_association
USING purged_user_ids
WHERE user_group_association.user_id = purged_user_ids.id
RETURNING user_group_association.user_id AS user_id,
user_group_association.id AS id),
deleted_ura_ids
AS (DELETE FROM user_role_association
USING role
WHERE role.id = user_role_association.role_id
AND role.type != 'private'
AND user_role_association.user_id IN
( SELECT id
FROM purged_user_ids)
RETURNING user_role_association.user_id AS user_id,
user_role_association.id AS id),
deleted_ua_ids
AS (DELETE FROM user_address
USING purged_user_ids
WHERE user_address.user_id = purged_user_ids.id
RETURNING user_address.user_id AS user_id,
user_address.id AS id),
user_events
AS (INSERT INTO cleanup_event_user_association
(create_time, cleanup_event_id, user_id)
SELECT NOW(), %(event_id)s, id
FROM purged_user_ids),
history_events
AS (INSERT INTO cleanup_event_history_association
(create_time, cleanup_event_id, history_id)
SELECT NOW(), %(event_id)s, id
FROM purged_history_ids),
hda_events
AS (INSERT INTO cleanup_event_hda_association
(create_time, cleanup_event_id, hda_id)
SELECT NOW(), %(event_id)s, id
FROM purged_hda_ids),
metadata_file_events
AS (INSERT INTO cleanup_event_metadata_file_association
(create_time, cleanup_event_id, metadata_file_id)
SELECT NOW(), %(event_id)s, id
FROM deleted_metadata_file_ids),
icda_events
AS (INSERT INTO cleanup_event_icda_association
(create_time, cleanup_event_id, icda_id)
SELECT NOW(), %(event_id)s, id
FROM deleted_icda_ids),
icda_hda_events
AS (INSERT INTO cleanup_event_hda_association
(create_time, cleanup_event_id, hda_id)
SELECT NOW(), %(event_id)s, hda_id
FROM deleted_icda_ids)
SELECT purged_user_ids.id AS purged_user_id,
purged_user_ids.id AS disk_accounting_user_id,
purged_history_ids.id AS purged_history_id,
purged_hda_ids.id AS purged_hda_id,
deleted_metadata_file_ids.id AS deleted_metadata_file_id,
deleted_metadata_file_ids.object_store_id AS object_store_id,
deleted_icda_ids.id AS deleted_icda_id,
deleted_icda_ids.hda_id AS deleted_icda_hda_id,
deleted_uga_ids.id AS deleted_uga_id,
deleted_ura_ids.id AS deleted_ura_id,
deleted_ua_ids.id AS deleted_ua_id
FROM purged_user_ids
LEFT OUTER JOIN purged_history_ids
ON purged_user_ids.id = purged_history_ids.user_id
LEFT OUTER JOIN purged_hda_ids
ON purged_history_ids.id = purged_hda_ids.history_id
LEFT OUTER JOIN deleted_metadata_file_ids
ON deleted_metadata_file_ids.hda_id = purged_hda_ids.id
LEFT OUTER JOIN deleted_icda_ids
ON deleted_icda_ids.hda_parent_id = purged_hda_ids.id
LEFT OUTER JOIN deleted_uga_ids
ON purged_user_ids.id = deleted_uga_ids.user_id
LEFT OUTER JOIN deleted_ura_ids
ON purged_user_ids.id = deleted_ura_ids.user_id
LEFT OUTER JOIN deleted_ua_ids
ON purged_user_ids.id = deleted_ua_ids.user_id
ORDER BY purged_user_ids.id
"""
cur = self._update(sql, {})
causals = (
('purged_user_id', 'purged_history_id'),
('purged_history_id', 'purged_hda_id'),
('purged_hda_id', 'deleted_metadata_file_id', 'object_store_id'),
('purged_hda_id', 'deleted_icda_id', 'deleted_icda_hda_id'),
('purged_user_id', 'deleted_uga_id'),
('purged_user_id', 'deleted_ura_id'),
('purged_user_id', 'deleted_ua_id'),
)
self._handle_results(cur, causals)
def delete_userless_histories(self):
"""
Mark deleted all "anonymous" Histories (not owned by a registered user) that are older than the specified number of days.
"""
log.info('Marking deleted all userless Histories older than %i days' % self.args.days)
sql = """
WITH deleted_history_ids
AS ( UPDATE history
@@ -380,18 +671,12 @@ class Cleanup(object):
(create_time, cleanup_event_id, history_id)
SELECT NOW(), %(event_id)s, id
FROM deleted_history_ids)
SELECT id
SELECT id AS deleted_history_id
FROM deleted_history_ids
ORDER BY id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked userless History deleted: %s' % tup[0])
self._close_logfile()
self._handle_results(cur)
def purge_error_hdas(self):
"""
@@ -399,7 +684,6 @@ class Cleanup(object):
number of days.
"""
log.info('Marking purged all error state HistoryDatasetAssociations older than %i days' % self.args.days)
sql = """
WITH purged_hda_ids
AS ( UPDATE history_dataset_association
@@ -415,30 +699,21 @@ class Cleanup(object):
(create_time, cleanup_event_id, hda_id)
SELECT NOW(), %(event_id)s, id
FROM purged_hda_ids)
SELECT purged_hda_ids.id,
history.user_id
SELECT purged_hda_ids.id AS purged_hda_id,
history.user_id AS disk_accounting_user_id
FROM purged_hda_ids
LEFT OUTER JOIN history
ON purged_hda_ids.history_id = history.id
ORDER BY purged_hda_ids.id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked HistoryDatasetAssociations purged: %s' % tup[0])
if tup[1] is not None and tup[1] not in self.disk_accounting_user_ids:
self.disk_accounting_user_ids.append(int(tup[1]))
self._close_logfile()
self._handle_results(cur)
def purge_hdas_of_purged_histories(self):
"""
Mark purged all HistoryDatasetAssociations in histories that are purged and older than the specified number of days.
"""
log.info('Marking purged all HistoryDatasetAssociations in purged Histories older than %i days' % self.args.days)
sql = """
WITH purged_hda_ids
AS ( UPDATE history_dataset_association
@@ -455,23 +730,15 @@ class Cleanup(object):
(create_time, cleanup_event_id, hda_id)
SELECT NOW(), %(event_id)s, id
FROM purged_hda_ids)
SELECT purged_hda_ids.id,
history.user_id
SELECT purged_hda_ids.id AS purged_hda_id,
history.user_id AS disk_accounting_user_id
FROM purged_hda_ids
LEFT OUTER JOIN history
ON purged_hda_ids.history_id = history.id
ORDER BY purged_hda_ids.id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked HistoryDatasetAssociations purged: %s' % tup[0])
if tup[1] is not None and tup[1] not in self.disk_accounting_user_ids:
self.disk_accounting_user_ids.append(int(tup[1]))
self._close_logfile()
self._handle_results(cur)
def purge_deleted_hdas(self):
"""
@@ -481,7 +748,6 @@ class Cleanup(object):
Mark purged all HistoryDatasetAssociations for which an ImplicitlyConvertedDatasetAssociation with matching hda_id is deleted in this step.
"""
log.info('Marking purged all deleted HistoryDatasetAssociations older than %i days' % self.args.days)
sql = """
WITH purged_hda_ids
AS ( UPDATE history_dataset_association
@@ -545,10 +811,7 @@ class Cleanup(object):
LEFT OUTER JOIN history
ON purged_hda_ids.history_id = history.id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked HistoryDatasetAssociations purged: %s' % tup[0])
@@ -556,7 +819,7 @@ class Cleanup(object):
self.disk_accounting_user_ids.append(int(tup[1]))
if tup[2] is not None:
self._log('Purge of HDA %s caused deletion of MetadataFile: %s in Object Store: %s' % (tup[0], tup[2], tup[3]))
self._remove_metadata_file(tup[2], tup[3], inspect.stack()[0][3])
self._remove_metadata_file(tup[2], tup[3], self.current_action)
if tup[4] is not None:
self._log('Purge of HDA %s caused deletion of ImplicitlyConvertedDatasetAssociation: %s and converted HistoryDatasetAssociation: %s' % (tup[0], tup[4], tup[5]))
self._close_logfile()
@@ -567,7 +830,6 @@ class Cleanup(object):
Mark purged all HistoryDatasetAssociations in Histories marked purged in this step (if not already purged).
"""
log.info('Marking purged all deleted histories that are older than the specified number of days.')
sql = """
WITH purged_history_ids
AS ( UPDATE history
@@ -646,10 +908,7 @@ class Cleanup(object):
ON deleted_icda_ids.hda_parent_id = purged_hda_ids.id
ORDER BY purged_history_ids.id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked History purged: %s' % tup[0])
@@ -659,7 +918,7 @@ class Cleanup(object):
self._log('Purge of History %s caused deletion of HistoryDatasetAssociation: %s' % (tup[0], tup[2]))
if tup[3] is not None:
self._log('Purge of HDA %s caused deletion of MetadataFile: %s in Object Store: %s' % (tup[1], tup[3], tup[4]))
self._remove_metadata_file(tup[3], tup[4], inspect.stack()[0][3])
self._remove_metadata_file(tup[3], tup[4], self.current_action)
if tup[5] is not None:
self._log('Purge of HDA %s caused deletion of ImplicitlyConvertedDatasetAssociation: %s and converted HistoryDatasetAssociation: %s' % (tup[1], tup[5], tup[6]))
self._close_logfile()
@@ -669,7 +928,6 @@ class Cleanup(object):
Mark deleted all Datasets that are derivative of JobExportHistoryArchives that are older than the specified number of days.
"""
log.info('Marking deleted all Datasets that are derivative of JobExportHistoryArchives that are older than the specified number of days.')
sql = """
WITH deleted_dataset_ids
AS ( UPDATE dataset
@@ -688,10 +946,7 @@ class Cleanup(object):
FROM deleted_dataset_ids
ORDER BY id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked Dataset deleted: %s' % tup[0])
@@ -702,7 +957,6 @@ class Cleanup(object):
Mark deleted all Datasets whose associations are all marked as deleted (LDDA) or purged (HDA) that are older than the specified number of days.
"""
log.info('Marking deleted all Datasets whose associations are all marked as deleted/purged that are older than the specified number of days.')
sql = """
WITH deleted_dataset_ids
AS ( UPDATE dataset
@@ -728,10 +982,7 @@ class Cleanup(object):
FROM deleted_dataset_ids
ORDER BY id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked Dataset deleted: %s' % tup[0])
@@ -742,7 +993,6 @@ class Cleanup(object):
Mark purged all Datasets marked deleted that are older than the specified number of days.
"""
log.info('Marking purged all Datasets marked deleted that are older than the specified number of days.')
sql = """
WITH purged_dataset_ids
AS ( UPDATE dataset
@@ -756,15 +1006,12 @@ class Cleanup(object):
(create_time, cleanup_event_id, dataset_id)
SELECT NOW(), %(event_id)s, id
FROM purged_dataset_ids)
SELECT id,
object_store_id
SELECT id AS purged_dataset_id,
object_store_id AS object_store_id
FROM purged_dataset_ids
ORDER BY id
"""
cur = self._update(sql, {})
self._flush()
self._open_logfile()
for tup in cur:
self._log('Marked Dataset purged: %s in Object Store: %s' % (tup[0], tup[1]))