Mforns has uploaded a new change for review. (
https://gerrit.wikimedia.org/r/359442 )
Change subject: [WIP] Modify EL purging script to not use limit/offset
......................................................................
[WIP] Modify EL purging script to not use limit/offset
Bug: T168071
Change-Id: Ic4c1e5a30ce41d8bc5ea3429f716d145118e3e65
---
M modules/role/files/mariadb/eventlogging_cleaner.py
1 file changed, 58 insertions(+), 87 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/operations/puppet
refs/changes/42/359442/1
diff --git a/modules/role/files/mariadb/eventlogging_cleaner.py
b/modules/role/files/mariadb/eventlogging_cleaner.py
index 4e536f0..f4fc85e 100644
--- a/modules/role/files/mariadb/eventlogging_cleaner.py
+++ b/modules/role/files/mariadb/eventlogging_cleaner.py
@@ -167,11 +167,12 @@
self.start = self.relative_ts(newer_than)
self.end = self.relative_ts(older_than)
self.batch_size = batch_size
+ self.max_interval_size = timedelta(days=30)
self.sleep_between_batches = sleep_between_batches
self.dry_run = dry_run
def relative_ts(self, days):
- return (self.reference_time -
timedelta(days=days)).strftime(DATE_FORMAT)
+ return self.reference_time - timedelta(days=days)
def purge(self, table):
"""
@@ -184,8 +185,8 @@
"LIMIT %(batch_size)s".format(table)
)
params = {
- 'start_ts': self.start,
- 'end_ts': self.end,
+ 'start_ts': self.start.strftime(DATE_FORMAT),
+ 'end_ts': self.end.strftime(DATE_FORMAT),
'batch_size': self.batch_size,
}
result = self.database.execute(command, params, dry_run=self.dry_run)
@@ -196,105 +197,75 @@
result = self.database.execute(command, params,
dry_run=self.dry_run)
time.sleep(self.sleep_between_batches)
- def _get_old_uuids(self, table, offset):
+ def get_interval_factor(self, numrows):
+ """The adapt factor tries to maintain the size of the update interval
+ close to the defined batch size. If the given table has lots of events
+ the update interval should be smaller, and if the table has less events
+ then the interval can be greater, so that we need less update queries.
+ The initial update interval is 1 minute, and then depending on the
+ number of updated rows, the adapt factor will multiply the interval
size
+ to make it closer to the batch size.
"""
- Return a list of uuids between self.start and self.end limiting
- the batch with an offset.
- """
- command = (
- "SELECT uuid from {} WHERE timestamp >= %(start_ts)s "
- "AND timestamp < %(end_ts)s LIMIT %(batch_size)s OFFSET %(offset)s"
- .format(table)
+ return min( # the min ensures that the factor is not too big
+ # batch_size / numrows should be 1 if the last update had
+ # a number of rows equal to batch_size. If the update was smaller,
+ # the factor should be proportionally greater, and viceversa.
+ self.batch_size / float(max(numrows, 1)), # max prevents division
by 0
+ 10
)
- params = {
- 'start_ts': self.start,
- 'end_ts': self.end,
- 'batch_size': self.batch_size,
- 'offset': offset,
- }
- result = self.database.execute(command, params, self.dry_run)
- if result['rows']:
- return [x[0] for x in result['rows']]
- else:
- return []
+
+ def interval_size_is_ok(self, interval_size, numrows):
+ return (
+ numrows < self.batch_size * 1.2 and (
+ numrows > self.batch_size * 0.8 or
+ interval_size == self.max_interval_size
+ )
+ )
+
+ def get_new_interval_size(self, table, interval_start, interval_size):
+ command_template = (
+ "SELECT count(*) FROM {}"
+ "WHERE timestamp >= {} AND timestamp < {{}}"
+ .format(table, interval_start.strftime(DATE_FORMAT))
+ )
+ interval_end = interval_start + interval_size
+ command = command_template.format(interval_end.strftime(DATE_FORMAT))
+ numrows = self.database.execute(command)['rows'][0][0]
+ while not self.interval_size_is_ok(interval_size, numrows):
+ interval_size = min(
+ interval_size * self.get_interval_factor(numrows),
+ self.max_interval_size
+ )
+ interval_end = interval_start + interval_size
+ command =
command_template.format(interval_end.strftime(DATE_FORMAT))
+ numrows = self.database.execute(command)['rows'][0][0]
+ return interval_size
def sanitize(self, table):
- """
- Set all the fields not in the whitelist (for a given table) to NULL.
- The schema_prefix is needed since the whitelist contains only
EventLogging
- schema/table prefixes.
- """
- # Get the table's whitelist prefix to retrieve the list of fields to
save
- # from the whitelist
- table_prefix = table.split('_')[0]
- # Sanity check
- if table_prefix not in self.whitelist:
- raise RuntimeError(
- "Sanitize has been called for table {}, but its "
- "prefix {} is not in the whitelist. Aborting as precautionary "
- "measure since this error condition might indicate a bug in
the code"
- .format(table, table_prefix)
- )
fields = self.database.get_table_fields(table)
- # Sanity check
- for field in self.whitelist[table_prefix]:
- if field not in fields:
- raise RuntimeError(
- "Sanitize has been called for table {}, but the field {} "
- "(belonging to the ones to keep/whitelist) is not part of "
- "the table fields. This probably indicates an error in the
"
- "whitelist, please re-check it. Aborting as precautionary "
- "measue to avoid unnecessary deletions."
- .format(table, field)
- )
+ # Get the schema name (without revision number)
+ table_prefix = table.split('_')[0]
fields_to_keep = self.whitelist[table_prefix] +
list(COMMON_PERSISTENT_FIELDS)
fields_to_purge = filter(lambda f: f not in fields_to_keep, fields)
-
values_string = ','.join([field + ' = NULL' for field in
fields_to_purge])
- offset = 0
- uuids_current_batch = self._get_old_uuids(table, offset)
command_template = (
"UPDATE {0} "
"SET {1} "
- "WHERE uuid IN ({{}})"
+ "WHERE timestamp >= {{}} AND timestamp < {{}}"
).format(table, values_string)
-
- while uuids_current_batch:
- uuids_no = len(uuids_current_batch)
- if uuids_no > self.batch_size:
- log.error("The number of uuids to sanitize {} is bigger "
- "than the batch size {}, this condition should not "
- "be possible, please review the code/data. "
- .format(str(uuids_no), str(self.batch_size)))
- raise RuntimeError('Sanitization stopped as precautionary
step.')
-
- uuids_current_batch_escaped = ["'" + x + "'" for x in
uuids_current_batch]
+ interval_start = self.start
+ interval_size = timedelta(minutes=1)
+ while interval_start < self.end:
+ interval_size = self.get_new_interval_size(table, interval_start,
interval_size)
+ interval_end = min(interval_start + interval_size, self.end)
result = self.database.execute(
- command_template.format(",".join(uuids_current_batch_escaped)),
+ command_template.format(
+ interval_start.strftime(DATE_FORMAT),
+ interval_end.strftime(DATE_FORMAT)
+ ),
dry_run=self.dry_run
)
- if result['numrows'] < uuids_no:
- log.warn("The number of uuids to sanitize {} is bigger "
- "than the number of updated rows in this batch {}. "
- "Data already sanitized?"
- .format(uuids_no, result['numrows']))
- elif result['numrows'] > uuids_no:
- log.error("The number of uuids to sanitize {} is lower "
- "than the number of updated rows in this batch {}. "
- "This is definitely an error in the code, please
review it."
- .format(uuids_no, result['numrows']))
- raise RuntimeError('Sanitization stopped as precautionary
step.')
-
- if uuids_no < self.batch_size:
- # Avoid an extra SQL query to the database if the number of
- # uuids returned are less than BATCH_SIZE, since this value
- # means that we have already reached the last batch of uuids
- # to sanitize.
- uuids_current_batch = []
- else:
- offset += self.batch_size
- uuids_current_batch = self._get_old_uuids(table, offset)
-
+ interval_start = interval_end
time.sleep(self.sleep_between_batches)
--
To view, visit https://gerrit.wikimedia.org/r/359442
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ic4c1e5a30ce41d8bc5ea3429f716d145118e3e65
Gerrit-PatchSet: 1
Gerrit-Project: operations/puppet
Gerrit-Branch: production
Gerrit-Owner: Mforns <mfo...@wikimedia.org>
_______________________________________________
MediaWiki-commits mailing list
MediaWiki-commits@lists.wikimedia.org
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
