diff --git a/scripts/cleanup_datasets/pgcleanup.py b/scripts/cleanup_datasets/pgcleanup.py index 01843ae5922..6ca012d7dfe 100755 --- a/scripts/cleanup_datasets/pgcleanup.py +++ b/scripts/cleanup_datasets/pgcleanup.py @@ -35,21 +35,46 @@ DEFAULT_LOG_DIR = os.path.join(galaxy_root, 'scripts', 'cleanup_datasets') log = logging.getLogger(__name__) -MetadataFile = namedtuple('MetadataFile', ['id', 'object_store_id']) -Dataset = namedtuple('Dataset', ['id', 'object_store_id']) + +## +## BASE CLASSES, etc. +## -class RemovesMetadataFiles(object): - pass - - -class RemovesDatasets(object): - pass +class LevelFormatter(logging.Formatter): + warn_fmt = "%(levelname)-5s %(funcName)s(): %(message)s" + def_fmt = "%(message)s" + def format(self, record): + if record.levelno > logging.INFO: + fmt = self.warn_fmt + else: + fmt = self.def_fmt + if hasattr(self, '_style'): # py3 + self._style._fmt = fmt + else: + self._fmt = fmt + return logging.Formatter.format(self, record) class Action(object): - update_time_sql = ", update_time = NOW()" + """Base class for all actions. + + When writing new actions, the following things happen automatically: + - _action_sql's format() method is used to replace ``{update_time_sql}`` and ``{force_retry_sql}`` + - an event is created and its id is added to the sql args with key ``event_id`` + - ``days`` is added to the sql args + - ``causals`` (see examples) allow logging what events triggered what other events + + The first column in the result set is considered the "primary" column upon which actions are being performed. + + See the mixins for additional features. + + Generally you should set at least ``_action_sql`` in subclasses (although it's possible to just override ``sql`` + directly.) + """ + update_time_sql = ", update_time = NOW() AT TIME ZONE 'utc'" force_retry_sql = " AND NOT purged" + primary_key = None causals = () _action_sql = "" _action_sql_args = {} @@ -58,46 +83,55 @@ class Action(object): @classmethod def name_c(cls): - name = [cls.__name__[0].lower()] - for c in cls.__name__[1:]: + # special case - for more complex stuff you can always implement name_c() on subclasses + clsname = cls.__name__.replace('HDA', 'Hda') + actname = [clsname[0].lower()] + for c in clsname[1:]: if c in string.ascii_uppercase: c = '_' + c.lower() - name.append(c) - return ''.join(name) + actname.append(c) + return ''.join(actname) @classmethod def doc_iter(cls): for line in cls.__doc__.splitlines(): yield line.replace(' ', '', 4) - def __init__(self, args): - self.__log_dir = args.log_dir - self.__dry_run = args.dry_run - self.__debug = args.debug - self.__update_time = args.update_time - self.__force_retry = args.force_retry - self.__days = args.days + def __init__(self, app): + self._log_dir = app.args.log_dir + self._dry_run = app.args.dry_run + self._debug = app.args.debug + self._update_time = app.args.update_time + self._force_retry = app.args.force_retry + self._days = app.args.days + self._config = app.config + self._update = app._update self.__log = None + self.__row_methods = [] + self.__post_methods = [] + self.__exit_methods = [] self._init() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): + for method in self.__exit_methods: + method() if self.__log is not None: self.__close_log() def __open_log(self): - logf = os.path.join(self.__log_dir, self.name + '.log') - if self.__dry_run: + logf = os.path.join(self._log_dir, self.name + '.log') + if self._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.__debug else logging.INFO) - f = logging.Formatter('%(message)s') - h.setFormatter(f) + h.setLevel(logging.DEBUG if self._debug else logging.INFO) + #f = logging.Formatter('%(message)s') + h.setFormatter(LevelFormatter()) l = logging.getLogger(self.name) l.addHandler(h) l.propagate = False @@ -110,6 +144,15 @@ class Action(object): self.log.info(m) self.__log = None + def _register_row_method(self, method): + self.__row_methods.append(method) + + def _register_post_method(self, method): + self.__post_methods.append(method) + + def _register_exit_method(self, method): + self.__exit_methods.append(method) + @property def log(self): if self.__log is None: @@ -126,14 +169,14 @@ class Action(object): 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.__update_time and not self.__force_retry: + if self._update_time and not self._force_retry: return self.update_time_sql return "" @property def _force_retry_sql(self): # a misnomer, the value of force_retry_sql is actually the SQL that should be used if *not* forcibly retrying - if not self.__force_retry: + if not self._force_retry: return self.force_retry_sql return "" @@ -148,7 +191,7 @@ class Action(object): def sql_args(self): args = self._action_sql_args.copy() if 'days' not in args: - args['days'] = self.__days + args['days'] = self._days return args def _collect_row_results(self, row, results, primary_key): @@ -162,90 +205,750 @@ class Action(object): 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, primary_key): for primary in sorted(results.keys()): - self._log('%s: %s' % (primary_key, primary)) + self.log.info('%s: %s' % (primary_key, primary)) for causal, s in zip(self.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)) - - # FIXME: can you do this better than with causals? + self.log.info('%s %s caused %s' % (causal[0], r[0], secondaries)) def handle_results(self, cur): - """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` - - XX :param causals: names of columns that caused actions to be performed on other columns - XX :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``. - """ results = {} - primary_key = None + primary_key = self.primary_key for row in cur: if primary_key is None: primary_key = row._fields[0] self._collect_row_results(row, results, primary_key) - self._collect_row_actions(row) + for method in self.__row_methods: + method(row) self._log_results(results, primary_key) + for method in self.__post_methods: + method() # subclasses can implement these def _init(self): + """Action initialization routines. + + Subclasses that implement ``_init()`` should call ``super()`` in it. + """ pass - def handle_row(self, row): - pass - def post_action(self): - pass +class RemovesObjects(object): + """Base class for mixins that remove objects from object stores. + """ + def _init(self): + self.objects_to_remove = set() + log.info('Initializing object store for action %s', self.name) + self.object_store = build_object_store_from_config(self._config) + self._register_row_method(self.collect_removed_object_info) + self._register_post_method(self.remove_objects) + self._register_exit_method(self.object_store.shutdown) + + def collect_removed_object_info(self, row): + object_id = getattr(row, self.id_column, None) + if object_id: + self.objects_to_remove.add(self.object_class(object_id, row.object_store_id)) + + def remove_objects(self): + for object_to_remove in sorted(self.objects_to_remove): + self.remove_object(object_to_remove) + + def remove_from_object_store(self, object_to_remove, object_store_kwargs, entire_dir=False, check_exists=False): + # only remove the "object store path" - if it's at an external_filename, that file will be untouched anyway + # (which is what we want) + loggers = (self.log, log) + try: + # TODO: get_filename() can do stuff like cache locally from S3, but we want the filename for logging + # purposes; this isn't ideal, object stores should probably have a noop way to get the "object store native + # identifier" which in the case of Disk would be the path + if not check_exists or self.object_store.exists(object_to_remove, **object_store_kwargs): + filename = self.object_store.get_filename(object_to_remove, **object_store_kwargs) + self.log.info('removing %s at: %s', object_to_remove, filename) + if not self._dry_run: + self.object_store.delete(object_to_remove, entire_dir=entire_dir, **object_store_kwargs) + except ObjectNotFound as e: + [l.warning('object store failure: %s: %s', object_to_remove, e) for l in loggers] + except Exception as e: + [l.error('delete failure: %s: %s', object_to_remove, e) for l in loggers] + + def remove_object(self, object_to_remove): + raise NotImplementedError() + + +## +## MIXINS +## + + +class RequiresDiskUsageRecalculation(object): + """Causes disk usage to be recalculated for affected users. + + To use, ensure your query returns a ``recalculate_disk_usage_user_id`` column. + """ + def _init(self): + self.__recalculate_disk_usage_user_ids = set() + self._register_row_action_method(self.collect_recalculate_disk_usage_user_id) + self._register_post_action_method(self.recalculate_disk_usage) + + def collect_recalculate_disk_usage_user_id(self, row): + if row.recalculate_disk_usage_user_id: + self.__recalculate_disk_usage_user_ids.add(row.recalculate_disk_usage_user_id) + + def recalculate_disk_usage(self): + """ + Any operation that purges a HistoryDatasetAssociation may require + updating a user's disk usage. Rather than attempt to resolve dataset + copies at purge-time, simply maintain a list of users that have had + HDAs purged, and update their usages once all updates are complete. + + This could probably be done more efficiently. + """ + log.info('Recalculating disk usage for users whose data were purged') + for user_id in sorted(self.__recalculate_disk_usage_user_ids): + # TODO: h.purged = false should be unnecessary once all hdas in purged histories are purged. + sql = """ + UPDATE galaxy_user + SET disk_usage = ( + SELECT COALESCE(SUM(total_size), 0) + FROM ( SELECT d.total_size + FROM history_dataset_association hda + JOIN history h ON h.id = hda.history_id + JOIN dataset d ON hda.dataset_id = d.id + WHERE h.user_id = %(user_id)s + AND h.purged = false + AND hda.purged = false + AND d.purged = false + AND d.id NOT IN (SELECT dataset_id + FROM library_dataset_dataset_association) + GROUP BY d.id) AS sizes) + WHERE id = %(user_id)s + RETURNING disk_usage; + """ + args = {'user_id': user_id} + cur = self._update(sql, args, add_event=False) + for row in cur: + # disk_usage might be None (e.g. user has purged all data) + self.log.info('recalculate_disk_usage user_id %i to %s bytes' % (user_id, row.disk_usage)) + + +class RemovesMetadataFiles(RemovesObjects): + """Causes MetadataFiles to be removed from the object store. + + To use, ensure your query returns ``deleted_metadata_file_id`` and ``object_store_id`` columns. + """ + object_class = namedtuple('MetadataFile', ['id', 'object_store_id']) + id_column = 'deleted_metadata_file_id' + + def remove_object(self, metadata_file): + self.remove_from_object_store( + metadata_file, + dict( + extra_dir='_metadata_files', + extra_dir_at_root=True, + alt_name="metadata_%d.dat" % metadata_file.id)) + + +class RemovesDatasets(RemovesObjects): + """Causes Datasets to be removed from the object store. + + To use, ensure your query returns ``purged_dataset_id`` and ``object_store_id`` columns. + """ + object_class = namedtuple('Dataset', ['id', 'object_store_id']) + id_column = 'purged_dataset_id' + + def remove_object(self, dataset): + self.remove_from_object_store(dataset, dict()) + self.remove_from_object_store( + dataset, + dict( + dir_only=True, + extra_dir="dataset_%d_files" % dataset.id), + entire_dir=True, + check_exists=True) + + +## +## ACTIONS +## + + +class UpdateHDAPurgedFlag(Action): + """ + The old cleanup script does not mark HistoryDatasetAssociations as purged when deleted Histories + are purged. This action can be used to rectify that situation. + """ + # update_time is intentionally left unmodified. + _action_sql = """ + WITH purged_hda_ids + AS ( UPDATE history_dataset_association + SET purged = true + FROM dataset + WHERE history_dataset_association.dataset_id = dataset.id + AND dataset.purged + AND NOT history_dataset_association.purged + RETURNING history_dataset_association.id), + 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) + SELECT id AS purged_hda_id + FROM purged_hda_ids + ORDER BY id; + """ + + +class DeleteUserlessHistories(Action): + """ + - Mark deleted all "anonymous" Histories (not owned by a registered user) that are older than + the specified number of days. + """ + _action_sql = """ + WITH deleted_history_ids + AS ( UPDATE history + SET deleted = true{update_time_sql} + WHERE user_id is null + AND NOT deleted + AND update_time < (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING id), + history_events + AS (INSERT INTO cleanup_event_history_association + (create_time, cleanup_event_id, history_id) + SELECT NOW() AT TIME ZONE 'utc', %(event_id)s, id + FROM deleted_history_ids) + SELECT id AS deleted_history_id + FROM deleted_history_ids + ORDER BY id + """ + + +class DeleteInactiveUsers(Action): + """ + - 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. + """ + force_retry_sql = " AND NOT deleted" + _action_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() AT TIME ZONE 'utc' - 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() AT TIME ZONE 'utc', %(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 + """ + causals = ( + ('purged_user_id', 'purged_job_id'), + ) + + +class PurgeDeletedUsers(RemovesMetadataFiles, Action): + """ + - 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. + """ + _action_sql = """ + WITH purged_user_ids + AS ( UPDATE galaxy_user + SET purged = true{update_time_sql} + WHERE deleted{force_retry_sql} + AND update_time < (NOW() AT TIME ZONE 'utc' - 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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(event_id)s, hda_id + FROM deleted_icda_ids) + SELECT purged_user_ids.id AS purged_user_id, + purged_user_ids.id AS zero_disk_usage_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 + """ + 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'), + ) + + def _init(self): + super(PurgeDeletedUsers, self)._init() + self.__zero_disk_usage_user_ids = set() + self._register_row_method(self.collect_zero_disk_usage_user_id) + self._register_post_method(self.zero_disk_usage) + + def collect_zero_disk_usage_user_id(self, row): + self.__zero_disk_usage_user_ids.add(row.zero_disk_usage_user_id) + + def zero_disk_usage(self): + log.info('Zeroing disk usage for users who were purged') + sql = """ + UPDATE galaxy_user + SET disk_usage = 0 + WHERE id IN %(user_ids)s + """ + user_ids = sorted(self.__zero_disk_usage_user_ids) + args = {'user_ids': tuple(user_ids)} + cur = self._update(sql, args, add_event=False) + self.log.info('zero_disk_usage user_ids: %s', ' '.join([str(i) for i in user_ids])) + + +class PurgeDeletedHDAs(RemovesMetadataFiles, RequiresDiskUsageRecalculation, Action): + """ + - Mark purged all HistoryDatasetAssociations currently marked deleted that are older than the + specified number of days. + - Mark deleted all MetadataFiles whose hda_id is purged in this step. + - Mark deleted all ImplicitlyConvertedDatasetAssociations whose hda_parent_id is purged in this + step. + - Mark purged all HistoryDatasetAssociations for which an ImplicitlyConvertedDatasetAssociation + with matching hda_id is deleted in this step. + """ + _action_sql = """ + WITH purged_hda_ids + AS ( UPDATE history_dataset_association + SET purged = true{update_time_sql} + WHERE deleted{force_retry_sql} + AND update_time < (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING id, + history_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), + hda_events + AS (INSERT INTO cleanup_event_hda_association + (create_time, cleanup_event_id, hda_id) + SELECT NOW() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(event_id)s, hda_id + FROM deleted_icda_ids) + SELECT purged_hda_ids.id AS purged_hda_id, + history.user_id AS recalculate_disk_usage_user_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 + FROM purged_hda_ids + 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 history + ON purged_hda_ids.history_id = history.id + ORDER BY purged_hda_ids.id + """ + causals = ( + ('purged_hda_id', 'deleted_metadata_file_id', 'object_store_id'), + ('purged_hda_id', 'deleted_icda_id', 'deleted_icda_hda_id'), + ) + + +class PurgeErrorHDAs(RequiresDiskUsageRecalculation, Action): + """ + - Mark purged all HistoryDatasetAssociations whose dataset_id is state = 'error' that are older + than the specified number of days. + """ + # FIXME: do we need to do anything with ICDAs and MFs here? probably, yes? + # force_retry? what use would force_retry actually be here? well for one, if ICDAs and MFs were being handled, it + # would cause them to be tried again, otherwise they'll be permenantly undeleted + # TODO: maybe instead of duplicating this over and over again and having it depend on the upsert, we should just + # have ICDA and MF actions? not sure i like that you'd have to run even more actions though + _action_sql = """ + WITH purged_hda_ids + AS ( UPDATE history_dataset_association + SET purged = true{update_time_sql} + FROM dataset + WHERE history_dataset_association.dataset_id = dataset.id + AND NOT history_dataset_association.purged + AND dataset.state = 'error' + AND history_dataset_association.update_time < (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING history_dataset_association.id as id, + history_dataset_association.history_id as history_id), + hda_events + AS (INSERT INTO cleanup_event_hda_association + (create_time, cleanup_event_id, hda_id) + SELECT NOW() AT TIME ZONE 'utc', %(event_id)s, id + FROM purged_hda_ids) + SELECT purged_hda_ids.id AS purged_hda_id, + history.user_id AS recalculate_disk_usage_user_id + FROM purged_hda_ids + LEFT OUTER JOIN history + ON purged_hda_ids.history_id = history.id + ORDER BY purged_hda_ids.id + """ + + +class PurgeHDAsOfPurgedHistories(RequiresDiskUsageRecalculation, Action): + """ + - Mark purged all HistoryDatasetAssociations in histories that are purged and older than the + specified number of days. + """ + # FIXME: ICDAs and MFs here as well yes? + _action_sql = """ + WITH purged_hda_ids + AS ( UPDATE history_dataset_association + SET purged = true{update_time_sql} + FROM history + WHERE history_dataset_association.history_id = history.id + AND NOT history_dataset_association.purged + AND history.purged + AND history.update_time < (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING history_dataset_association.id as id, + history_dataset_association.history_id as history_id), + hda_events + AS (INSERT INTO cleanup_event_hda_association + (create_time, cleanup_event_id, hda_id) + SELECT NOW() AT TIME ZONE 'utc', %(event_id)s, id + FROM purged_hda_ids) + SELECT purged_hda_ids.id AS purged_hda_id, + history.user_id AS recalculate_disk_usage_user_id + FROM purged_hda_ids + LEFT OUTER JOIN history + ON purged_hda_ids.history_id = history.id + ORDER BY purged_hda_ids.id + """ + + +class PurgeDeletedHistories(RequiresDiskUsageRecalculation, Action): + """ + - Mark purged all Histories marked deleted that are older than the specified number of days. + - Mark purged all HistoryDatasetAssociations in Histories marked purged in this step (if not + already purged). + """ + _action_sql = """ + WITH purged_history_ids + AS ( UPDATE history + SET purged = true{update_time_sql} + WHERE deleted{force_retry_sql} + AND update_time < (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING id, + user_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), + history_events + AS (INSERT INTO cleanup_event_history_association + (create_time, cleanup_event_id, history_id) + SELECT NOW() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(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() AT TIME ZONE 'utc', %(event_id)s, hda_id + FROM deleted_icda_ids) + SELECT purged_history_ids.id AS purged_history_id, + purged_history_ids.user_id AS recalculate_disk_usage_user_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 + FROM purged_history_ids + 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 + ORDER BY purged_history_ids.id + """ + causals = ( + ('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'), + ) + + +class DeleteExportedHistories(Action): + """ + - Mark deleted all Datasets that are derivative of JobExportHistoryArchives that are older than + the specified number of days. + """ + _action_sql = """ + WITH deleted_dataset_ids + AS ( UPDATE dataset + SET deleted = true{update_time_sql} + FROM job_export_history_archive + WHERE job_export_history_archive.dataset_id = dataset.id + AND NOT deleted + AND dataset.update_time <= (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING dataset.id), + dataset_events + AS (INSERT INTO cleanup_event_dataset_association + (create_time, cleanup_event_id, dataset_id) + SELECT NOW() AT TIME ZONE 'utc', %(event_id)s, id + FROM deleted_dataset_ids) + SELECT id AS deleted_dataset_id + FROM deleted_dataset_ids + ORDER BY id + """ + + +class DeleteDatasets(Action): + """ + - Mark deleted all Datasets whose associations are all marked as deleted (LDDA) or purged (HDA) + that are older than the specified number of days. + """ + # FIXME: should we do anything with JEHAs and other dataset associations here? + _action_sql = """ + WITH deleted_dataset_ids + AS ( UPDATE dataset + SET deleted = true{update_time_sql} + WHERE NOT deleted + AND NOT EXISTS + (SELECT true + FROM library_dataset_dataset_association + WHERE (NOT deleted + OR update_time >= (NOW() AT TIME ZONE 'utc' - interval '%(days)s days')) + AND dataset.id = dataset_id) + AND NOT EXISTS + (SELECT true + FROM history_dataset_association + WHERE (NOT purged + OR update_time >= (NOW() AT TIME ZONE 'utc' - interval '%(days)s days')) + AND dataset.id = dataset_id) + RETURNING id), + dataset_events + AS (INSERT INTO cleanup_event_dataset_association + (create_time, cleanup_event_id, dataset_id) + SELECT NOW() AT TIME ZONE 'utc', %(event_id)s, id + FROM deleted_dataset_ids) + SELECT id + FROM deleted_dataset_ids + ORDER BY id + """ class PurgeDatasets(RemovesDatasets, Action): """ - Mark purged all Datasets marked deleted that are older than the specified number of days. + - Mark purged all Datasets marked deleted that are older than the specified number of days. """ - - sql = """ - WITH purged_dataset_ids - AS ( UPDATE dataset - SET purged = true{update_time_sql} - WHERE deleted{force_retry_sql} - AND update_time < (NOW() - interval '%(days)s days') - RETURNING id, - object_store_id), - dataset_events - AS (INSERT INTO cleanup_event_dataset_association - (create_time, cleanup_event_id, dataset_id) - SELECT NOW(), %(event_id)s, id - FROM purged_dataset_ids) - SELECT id AS purged_dataset_id, - object_store_id AS object_store_id - FROM purged_dataset_ids - ORDER BY id + _action_sql = """ + WITH purged_dataset_ids + AS ( UPDATE dataset + SET purged = true{update_time_sql} + WHERE deleted{force_retry_sql} + AND update_time < (NOW() AT TIME ZONE 'utc' - interval '%(days)s days') + RETURNING id, + object_store_id), + dataset_events + AS (INSERT INTO cleanup_event_dataset_association + (create_time, cleanup_event_id, dataset_id) + SELECT NOW() AT TIME ZONE 'utc', %(event_id)s, id + FROM purged_dataset_ids) + SELECT id AS purged_dataset_id, + object_store_id AS object_store_id + FROM purged_dataset_ids + ORDER BY id """ - 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): @@ -253,19 +956,19 @@ class Cleanup(object): self.config = None self.__conn = None self.__actions = None - 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.__current_action = None self.__parse_args() self.__setup_logging() self.__validate_actions() self.__load_config() - self.__load_object_store() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.__conn is not None: + self.__conn.close() @property def actions(self): @@ -285,6 +988,10 @@ class Cleanup(object): args.update(url.query) assert url.get_dialect().name == 'postgresql', 'This script can only be used with PostgreSQL.' self.__conn = psycopg2.connect(cursor_factory=NamedTupleCursor, **args) + # TODO: is this per session or cursor? + if self.args.work_mem is not None: + log.info('Setting work_mem to %s' % self.args.work_mem) + self._conn.cursor().execute('SET work_mem TO %s', (self.args.work_mem,)) return self.__conn def __parse_args(self): @@ -346,10 +1053,9 @@ class Cleanup(object): def __setup_logging(self): format = "%(asctime)s %(levelname)-5s %(funcName)s(): %(message)s" - if self.args.debug: - logging.basicConfig(level=logging.DEBUG, format=format) - else: - logging.basicConfig(level=logging.INFO, format=format) + logging.basicConfig( + level=logging.DEBUG if self.args.debug else logging.INFO, + format="%(asctime)s %(levelname)-5s %(funcName)s(): %(message)s") def __validate_actions(self): ok = True @@ -365,88 +1071,60 @@ class Cleanup(object): app_properties = app_properties_from_args(self.args) self.config = galaxy.config.Configuration(**app_properties) - def __load_object_store(self): - self.object_store = build_object_store_from_config(self.config) + def _dry_run_event(self): + cur = self.conn.cursor() + sql = "SELECT MAX(id) FROM cleanup_event;" + cur.execute(sql) + max_id = cur.fetchone()[0] + if max_id is None: + # there has to be at least one event in the table, if there are none just create a fake one. + sql = "INSERT INTO cleanup_event (create_time, message) VALUES (NOW(), 'dry_run_event') RETURNING id;" + cur.execute(sql) + max_id = cur.fetchone()[0] + self.conn.commit() + log.info("An event must exist for the subsequent query to succeed, so a dummy event has been created") + else: + log.info("Not executing event creation (increments sequence even when rolling back), using an old " + "event ID (%i) for dry run" % max_id) + return max_id - def run(self): - for name in self.args.actions: - cls = self.actions[name] - log.info("Running action '%s':", name) - map(log.info, cls.doc_iter()) - self.current_action = name - with cls(self.args) as action: - self._run_action(action) - #self.__getattribute__(name)() - log.info('Finished %s' % name) - if self.disk_accounting_user_ids: - self.update_user_disk_usage() - - def _run_action(self, action): - self._update(action.sql, action.sql_args, force_retry_sql=action.force_retry_sql) + def _execute(self, sql, args): + cur = self.conn.cursor() + log.debug("SQL is: %s", cur.mogrify(sql, args)) + log.info("Executing SQL") + cur.execute(sql, args) + log.info('Database status: %s', cur.statusmessage) + return cur def _create_event(self, message=None): """ Create a new event in the cleanup_event table. """ - - if message is None: - message = self.current_action + if self.args.dry_run: + return self._dry_run_event() sql = """ INSERT INTO cleanup_event (create_time, message) - VALUES (NOW(), %(message)s) + VALUES (NOW() AT TIME ZONE 'utc', %(message)s) RETURNING id; """ - + message = message or self.__current_action args = {'message': message} - - log.debug("SQL is: %s", sql % args_for_print(args)) - - cur = self.conn.cursor() - - if self.args.dry_run: - sql = "SELECT MAX(id) FROM cleanup_event;" - cur.execute(sql) - max_id = cur.fetchone()[0] - if max_id is None: - # there has to be at least one event in the table, if there are none just create a fake one. - sql = "INSERT INTO cleanup_event (create_time, message) VALUES (NOW(), 'dry_run_event') RETURNING id;" - cur.execute(sql) - max_id = cur.fetchone()[0] - self.conn.commit() - log.info("An event must exist for the subsequent query to succeed, so a dummy event has been created") - else: - log.info("Not executing event creation (increments sequence even when rolling back), using an old event ID (%i) for dry run" % max_id) - return max_id - - log.info("Executing SQL") - cur.execute(sql, args) - log.info('Database status: %s' % cur.statusmessage) - - return cur.fetchone()[0] + event_id = self._execute(sql, args).fetchone()[0] + log.info("Created event %s for action: %s", event_id, self.__current_action) + return event_id def _update(self, sql, args, add_event=True, event_message=None): - if add_event and isinstance(args, dict) and 'event_id' not in args: + if add_event and 'event_id' not in args: args['event_id'] = self._create_event(message=event_message) - - if args is not None: - log.debug('SQL is: %s', sql % args_for_print(args)) + cur = self._execute(sql, args) + if cur.rowcount <= 0: + log.info("Update resulted in no changes, rolling back transaction") + self.conn.rollback() else: - log.debug('SQL is: %s', sql) - - cur = self.conn.cursor() - - if self.args.work_mem is not None: - log.info('Setting work_mem to %s' % self.args.work_mem) - cur.execute('SET work_mem TO %s', (self.args.work_mem,)) - - log.info('Executing SQL') - cur.execute(sql, args) - log.info('Database status: %s' % cur.statusmessage) - log.info('Flushing transaction') - self._flush() - + log.info('Flushing transaction') + self._flush() return cur def _flush(self): @@ -457,694 +1135,24 @@ class Cleanup(object): self.conn.commit() log.info("All changes committed") - def _remove_metadata_file(self, id, object_store_id): - metadata_file = MetadataFile(id=id, object_store_id=object_store_id) + def _run_action(self, action): + cur = self._update(action.sql, action.sql_args) + action.handle_results(cur) - 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) - except (ObjectNotFound, AttributeError) as e: - log.error('Unable to get MetadataFile %s filename: %s' % (id, e)) - return - - if not self.args.dry_run: - try: - os.unlink(filename) - except Exception as e: - self._log('Removal of %s failed with error: %s' % (filename, e)) - - def update_user_disk_usage(self): - """ - Any operation that purges a HistoryDatasetAssociation may require - updating a user's disk usage. Rather than attempt to resolve dataset - copies at purge-time, simply maintain a list of users that have had - HDAs purged, and update their usages once all updates are complete. - - This could probably be done more efficiently. - """ - log.info('Recalculating disk usage for users whose data were purged') - - 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 = """ - UPDATE galaxy_user - SET disk_usage = (SELECT COALESCE(SUM(total_size), 0) - FROM ( SELECT d.total_size - FROM history_dataset_association hda - JOIN history h ON h.id = hda.history_id - JOIN dataset d ON hda.dataset_id = d.id - WHERE h.user_id = %(user_id)s - AND h.purged = false - AND hda.purged = false - AND d.purged = false - AND d.id NOT IN (SELECT dataset_id - FROM library_dataset_dataset_association) - GROUP BY d.id) sizes) - WHERE id = %(user_id)s - RETURNING disk_usage; - """ - - args = {'user_id': user_id} - cur = self._update(sql, args, add_event=False) - - for tup in cur: - # disk_usage might be None (e.g. user has purged all data) - 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() - if self.action_logger is not None: - m = ('==== Log closed at shutdown: %s ' % datetime.datetime.now().isoformat()).ljust(72, '=') - self.action_logger.info(m) - - # - # actions - # - - # the _update method automatically: - # - calls the sql argument's (string) format() method to replace '{update_time_sql}' and '{force_retry_sql}' - # - creates an event and adds its id to args with key `event_id` (if args is a dict) - # - adds `days` to args (if args is a dict) - - def update_hda_purged_flag(self): - """ - 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 - AS ( UPDATE history_dataset_association - SET purged = true - FROM dataset - WHERE history_dataset_association.dataset_id = dataset.id - AND dataset.purged - AND NOT history_dataset_association.purged - RETURNING history_dataset_association.id), - 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) - SELECT id AS purged_hda_id - FROM purged_hda_ids - ORDER BY id; - """ - cur = self._update(sql, {}) - self._handle_results(cur) - - 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 - SET deleted = true{update_time_sql} - WHERE user_id is null - AND NOT deleted - AND update_time < (NOW() - interval '%(days)s days') - RETURNING id), - history_events - AS (INSERT INTO cleanup_event_history_association - (create_time, cleanup_event_id, history_id) - SELECT NOW(), %(event_id)s, id - FROM deleted_history_ids) - SELECT id AS deleted_history_id - FROM deleted_history_ids - ORDER BY id - """ - cur = self._update(sql, {}) - self._handle_results(cur) - - def purge_error_hdas(self): - """ - Mark purged all HistoryDatasetAssociations whose dataset_id is state = 'error' that are older than the specified - 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 - SET purged = true{update_time_sql} - FROM dataset - WHERE history_dataset_association.dataset_id = dataset.id - AND dataset.state = 'error' - AND history_dataset_association.update_time < (NOW() - interval '%(days)s days') - RETURNING history_dataset_association.id as id, - history_dataset_association.history_id as history_id), - 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) - 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._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 - SET purged = true{update_time_sql} - FROM history - WHERE history_dataset_association.history_id = history.id - AND NOT history_dataset_association.purged - AND history.purged - AND history.update_time < (NOW() - interval '%(days)s days') - RETURNING history_dataset_association.id as id, - history_dataset_association.history_id as history_id), - 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) - 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._handle_results(cur) - - def purge_deleted_hdas(self): - """ - Mark purged all HistoryDatasetAssociations currently marked deleted that are older than the specified number of days. - Mark deleted all MetadataFiles whose hda_id is purged in this step. - Mark deleted all ImplicitlyConvertedDatasetAssociations whose hda_parent_id is purged in this step. - 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 - SET purged = true{update_time_sql} - WHERE deleted{force_retry_sql} - AND update_time < (NOW() - interval '%(days)s days') - RETURNING id, - history_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), - 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_hda_ids.id, - history.user_id, - deleted_metadata_file_ids.id, - deleted_metadata_file_ids.object_store_id, - deleted_icda_ids.id, - deleted_icda_ids.hda_id - FROM purged_hda_ids - 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 history - ON purged_hda_ids.history_id = history.id - """ - cur = self._update(sql, {}) - 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])) - 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], 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() - - def purge_deleted_histories(self): - """ - Mark purged all Histories marked deleted that are older than the specified number of days. - 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 - SET purged = true{update_time_sql} - WHERE deleted{force_retry_sql} - AND update_time < (NOW() - interval '%(days)s days') - RETURNING id, - user_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), - 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_history_ids.id, - purged_history_ids.user_id, - purged_hda_ids.id, - deleted_metadata_file_ids.id, - deleted_metadata_file_ids.object_store_id, - deleted_icda_ids.id, - deleted_icda_ids.hda_id - FROM purged_history_ids - 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 - ORDER BY purged_history_ids.id - """ - cur = self._update(sql, {}) - self._open_logfile() - for tup in cur: - self._log('Marked History 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])) - if tup[2] is not None: - 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], 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() - - def delete_exported_histories(self): - """ - 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 - SET deleted = true{update_time_sql} - FROM job_export_history_archive - WHERE job_export_history_archive.dataset_id = dataset.id - AND NOT deleted - AND dataset.update_time <= (NOW() - interval '%(days)s days') - RETURNING dataset.id), - dataset_events - AS (INSERT INTO cleanup_event_dataset_association - (create_time, cleanup_event_id, dataset_id) - SELECT NOW(), %(event_id)s, id - FROM deleted_dataset_ids) - SELECT id - FROM deleted_dataset_ids - ORDER BY id - """ - cur = self._update(sql, {}) - self._open_logfile() - for tup in cur: - self._log('Marked Dataset deleted: %s' % tup[0]) - self._close_logfile() - - def delete_datasets(self): - """ - 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 - SET deleted = true{update_time_sql} - WHERE NOT deleted - AND NOT EXISTS (SELECT true - FROM library_dataset_dataset_association - WHERE (NOT deleted - OR update_time >= (NOW() - interval '%(days)s days')) - AND dataset.id = dataset_id) - AND NOT EXISTS (SELECT true - FROM history_dataset_association - WHERE (NOT purged - OR update_time >= (NOW() - interval '%(days)s days')) - AND dataset.id = dataset_id) - RETURNING id), - dataset_events - AS (INSERT INTO cleanup_event_dataset_association - (create_time, cleanup_event_id, dataset_id) - SELECT NOW(), %(event_id)s, id - FROM deleted_dataset_ids) - SELECT id - FROM deleted_dataset_ids - ORDER BY id - """ - cur = self._update(sql, {}) - self._open_logfile() - for tup in cur: - self._log('Marked Dataset deleted: %s' % tup[0]) - self._close_logfile() - - def purge_datasets(self): - log.info('Marking purged all Datasets marked deleted that are older than the specified number of days.') - sql = """ - """ - cur = self._update(sql, {}) - self._open_logfile() - for tup in cur: - self._log('Marked Dataset purged: %s in Object Store: %s' % (tup[0], tup[1])) - - # always try to remove the "object store path" - if it's at an external_filename, that file will be untouched anyway (which is what we want) - dataset = Dataset(id=tup[0], object_store_id=tup[1]) - try: - filename = self.object_store.get_filename(dataset) - except (ObjectNotFound, AttributeError) as e: - log.error('Unable to get Dataset %s filename: %s' % (tup[0], e)) - continue - - try: - extra_files_dir = self.object_store.get_filename(dataset, dir_only=True, extra_dir="dataset_%d_files" % tup[0]) - except (ObjectNotFound, AttributeError): - extra_files_dir = None - - # don't check for existence of the dataset, it should exist - self._log('Removing from disk: %s' % filename) - if not self.args.dry_run: - try: - os.unlink(filename) - except Exception as e: - self._log('Removal of %s failed with error: %s' % (filename, e)) - - # extra_files_dir is optional so it's checked first - if extra_files_dir is not None and os.path.exists(extra_files_dir): - self._log('Removing from disk: %s' % extra_files_dir) - if not self.args.dry_run: - try: - shutil.rmtree(extra_files_dir) - except Exception as e: - self._log('Removal of %s failed with error: %s' % (extra_files_dir, e)) - - self._close_logfile() - - -def quote_for_print(v): - if isinstance(v, string_types): - return "'" + v.replace("'", "\\'") + "'" - else: - return v - - -def args_for_print(args): - if isinstance(args, dict): - r = {} - for k, v in args.items(): - r[k] = quote_for_print(v) - elif isinstance(args, (tuple, list)): - r = [] - for v in args: - r.append(quote_for_print(v)) - if isinstance(args, tuple): - r = tuple(r) - else: - r = args - return r + def run(self): + for name in self.args.actions: + cls = self.actions[name] + log.info("Running action '%s':", name) + map(log.info, cls.doc_iter()) + self.__current_action = name + with cls(self) as action: + self._run_action(action) + log.info('Finished %s' % name) if __name__ == '__main__': - cleanup = Cleanup() - try: - cleanup.run() - except Exception: - log.exception('Caught exception in run sequence:') - cleanup._shutdown() + with Cleanup() as app: + try: + app.run() + except Exception: + log.exception('Caught exception in run sequence:')