219 lines
8.8 KiB
Python
219 lines
8.8 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
"""One-time migration: remote_file.remote_mtime from local-time epochs to UTC.
|
||
|
|
|
||
|
|
``puller_fits.py`` parsed each directory listing's timestamp with ``strptime`` and
|
||
|
|
then fed it to ``time.mktime``, which reads a naive datetime as *local* time. The
|
||
|
|
listings are UTC, so every recorded mtime came out shifted by this machine's UTC
|
||
|
|
offset -- 4 or 5 hours here, depending on the date's DST.
|
||
|
|
|
||
|
|
That was self-consistent while the puller kept making the same mistake on the same
|
||
|
|
machine, which is why nothing was ever re-downloaded because of it. It stopped
|
||
|
|
being self-consistent the moment the parser was fixed to read UTC, and it was never
|
||
|
|
safe: the archive has already moved hosts once, and on a box east of UTC the shift
|
||
|
|
reverses sign, every stored mtime reads as older than the listing, and the entire
|
||
|
|
archive looks like it needs re-fetching.
|
||
|
|
|
||
|
|
The transform recovers the wall-clock string the listing showed -- which is exactly
|
||
|
|
what ``fromtimestamp`` returns for a value ``mktime`` produced -- and re-reads it as
|
||
|
|
UTC::
|
||
|
|
|
||
|
|
stored 1673172480 -> wall '2023-01-08 05:08' -> 1673154480
|
||
|
|
|
||
|
|
Run this with the puller stopped. It rewrites every row in the table, and
|
||
|
|
``puller_fits.py`` warns on every start until the ``remote_mtime_utc`` marker is
|
||
|
|
set, so a half-migrated index cannot pass unnoticed.
|
||
|
|
|
||
|
|
migrate_mtime_utc.py # dry run: report the shift, change nothing
|
||
|
|
migrate_mtime_utc.py --apply # rewrite the rows
|
||
|
|
"""
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import collections
|
||
|
|
import csv
|
||
|
|
import datetime
|
||
|
|
import gzip
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
|
||
|
|
from suvi import db, paths
|
||
|
|
|
||
|
|
MARKER = "remote_mtime_utc"
|
||
|
|
#: How far the last run got, in url order. Written in the same transaction as the
|
||
|
|
#: batch it describes, so an interrupted migration resumes exactly where it stopped
|
||
|
|
#: rather than shifting an already-converted prefix a second time.
|
||
|
|
CURSOR = MARKER + ":cursor"
|
||
|
|
|
||
|
|
|
||
|
|
def to_utc(stored):
|
||
|
|
"""Re-read a local-time epoch as the UTC instant its wall clock named.
|
||
|
|
|
||
|
|
Returns (utc_epoch, ambiguous). `ambiguous` marks the one case the old value
|
||
|
|
cannot be inverted: on the day the local clock springs forward, the skipped
|
||
|
|
hour and the hour after it both map to this epoch, so a listing that read
|
||
|
|
02:30 and one that read 03:30 are now indistinguishable. We take the later
|
||
|
|
reading, which is the safe guess -- a row that was really 02:30 ends up an hour
|
||
|
|
*ahead* of the truth, so the next listing looks older and the file is skipped
|
||
|
|
rather than re-downloaded. Roughly one row in ten thousand.
|
||
|
|
"""
|
||
|
|
wall = datetime.datetime.fromtimestamp(stored)
|
||
|
|
hour_earlier = wall - datetime.timedelta(hours=1)
|
||
|
|
ambiguous = time.mktime(hour_earlier.timetuple()) == stored
|
||
|
|
return wall.replace(tzinfo=datetime.timezone.utc).timestamp(), ambiguous
|
||
|
|
|
||
|
|
|
||
|
|
def iter_batches(conn, batch, last=""):
|
||
|
|
"""Walk the table in url order from `last`, yielding [(url, remote_mtime), ...].
|
||
|
|
|
||
|
|
Keyset pagination rather than OFFSET: the table is WITHOUT ROWID with url as
|
||
|
|
its primary key, so this walks the index once instead of rescanning, and the
|
||
|
|
updates it drives never move a row's key.
|
||
|
|
"""
|
||
|
|
while True:
|
||
|
|
rows = conn.execute(
|
||
|
|
"SELECT url, remote_mtime FROM remote_file WHERE url > ? ORDER BY url LIMIT ?",
|
||
|
|
(last, batch),
|
||
|
|
).fetchall()
|
||
|
|
if not rows:
|
||
|
|
return
|
||
|
|
yield [(r["url"], r["remote_mtime"]) for r in rows]
|
||
|
|
last = rows[-1]["url"]
|
||
|
|
|
||
|
|
|
||
|
|
def write_backup(conn, path, batch=50000):
|
||
|
|
"""Dump (url, remote_mtime) before anything is rewritten.
|
||
|
|
|
||
|
|
The shift is invertible and the run is resumable, so this is belt and braces --
|
||
|
|
but remote_file is the only record of what has been downloaded, and a column
|
||
|
|
rewritten across every row of it is worth being able to put back verbatim.
|
||
|
|
Restoring is a plain executemany against this file.
|
||
|
|
"""
|
||
|
|
started = time.time()
|
||
|
|
rows = 0
|
||
|
|
with gzip.open(path, "wt", newline="") as handle:
|
||
|
|
writer = csv.writer(handle)
|
||
|
|
writer.writerow(["url", "remote_mtime"])
|
||
|
|
for chunk in iter_batches(conn, batch):
|
||
|
|
writer.writerows(chunk)
|
||
|
|
rows += len(chunk)
|
||
|
|
print(f" backing up {rows}...", end="\r", flush=True)
|
||
|
|
print(f" backed up {rows} rows to {path} "
|
||
|
|
f"({os.path.getsize(path) / 1024 / 1024:.0f} MB) in {time.time() - started:.1f}s")
|
||
|
|
return rows
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv=None):
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__,
|
||
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
|
|
parser.add_argument("--db", default=None, help="index to migrate (default: $SUVI_DB)")
|
||
|
|
parser.add_argument("--apply", action="store_true", help="rewrite rows (default: dry run)")
|
||
|
|
parser.add_argument("--batch", type=int, default=50000)
|
||
|
|
parser.add_argument("--force", action="store_true",
|
||
|
|
help="re-run even though the migration is marked done")
|
||
|
|
parser.add_argument("--backup", default=None,
|
||
|
|
help="where to dump the old mtimes (default: beside the index)")
|
||
|
|
parser.add_argument("--no-backup", action="store_true",
|
||
|
|
help="skip the pre-flight dump of the old mtimes")
|
||
|
|
args = parser.parse_args(argv)
|
||
|
|
|
||
|
|
target = args.db or paths.default_db_path()
|
||
|
|
conn = db.connect(target)
|
||
|
|
|
||
|
|
done = db.get_meta(conn, MARKER)
|
||
|
|
if done == "done" and not args.force:
|
||
|
|
print(f"{target} is already marked migrated ({MARKER}=done); nothing to do.")
|
||
|
|
print("Applying the shift twice would corrupt every record. Pass --force only")
|
||
|
|
print("if you are certain the marker is wrong.")
|
||
|
|
conn.close()
|
||
|
|
return 0
|
||
|
|
|
||
|
|
total = conn.execute("SELECT count(*) c FROM remote_file").fetchone()["c"]
|
||
|
|
print(f"{target}: {total} remote_file rows")
|
||
|
|
if not total:
|
||
|
|
conn.close()
|
||
|
|
return 0
|
||
|
|
|
||
|
|
resume_from = db.get_meta(conn, CURSOR) or ""
|
||
|
|
to_process = total
|
||
|
|
if resume_from:
|
||
|
|
to_process = conn.execute(
|
||
|
|
"SELECT count(*) c FROM remote_file WHERE url > ?", (resume_from,)
|
||
|
|
).fetchone()["c"]
|
||
|
|
print(f" resuming an interrupted run: {total - to_process} rows already converted,"
|
||
|
|
f" {to_process} to go")
|
||
|
|
|
||
|
|
if args.apply and not args.no_backup:
|
||
|
|
backup = args.backup or os.path.join(
|
||
|
|
os.path.dirname(os.path.abspath(target)),
|
||
|
|
f"remote_mtime_pre_utc_{time.strftime('%Y%m%dT%H%M%S')}.csv.gz",
|
||
|
|
)
|
||
|
|
write_backup(conn, backup, args.batch)
|
||
|
|
|
||
|
|
shifts = collections.Counter()
|
||
|
|
ambiguous = []
|
||
|
|
samples = []
|
||
|
|
seen = written = 0
|
||
|
|
started = time.time()
|
||
|
|
|
||
|
|
for rows in iter_batches(conn, args.batch, resume_from):
|
||
|
|
updates = []
|
||
|
|
for url, stored in rows:
|
||
|
|
utc, is_ambiguous = to_utc(stored)
|
||
|
|
shifts[int(stored - utc)] += 1
|
||
|
|
if is_ambiguous and len(ambiguous) < 20:
|
||
|
|
ambiguous.append(url)
|
||
|
|
if len(samples) < 3:
|
||
|
|
samples.append((url, stored, utc))
|
||
|
|
updates.append((utc, url))
|
||
|
|
seen += len(rows)
|
||
|
|
|
||
|
|
if args.apply:
|
||
|
|
# The cursor rides in the same transaction as the rows it covers, so a
|
||
|
|
# kill at any point leaves the two consistent and the resume exact.
|
||
|
|
conn.executemany(
|
||
|
|
"UPDATE remote_file SET remote_mtime = ? WHERE url = ?", updates
|
||
|
|
)
|
||
|
|
db.set_meta(conn, CURSOR, rows[-1][0])
|
||
|
|
conn.commit()
|
||
|
|
written += len(updates)
|
||
|
|
print(f" {written}/{to_process}", end="\r", flush=True)
|
||
|
|
|
||
|
|
print(f" scanned {seen} rows in {time.time() - started:.1f}s" + " " * 20)
|
||
|
|
print(" shift applied (stored - utc), in seconds:")
|
||
|
|
for shift, count in shifts.most_common(10):
|
||
|
|
print(f" {shift:+7d} ({shift / 3600:+.1f} h): {count} rows")
|
||
|
|
if len(shifts) > 10:
|
||
|
|
print(f" ... and {len(shifts) - 10} other shifts")
|
||
|
|
if ambiguous:
|
||
|
|
print(f" {len(ambiguous)}{'+' if len(ambiguous) == 20 else ''} rows sit in a spring-forward"
|
||
|
|
" hour and may land an hour ahead (they will be skipped, not re-fetched),")
|
||
|
|
print(f" e.g. {ambiguous[0]}")
|
||
|
|
for url, stored, utc in samples:
|
||
|
|
print(f" e.g. {url}")
|
||
|
|
print(f" {stored:.0f} -> {utc:.0f}"
|
||
|
|
f" ({datetime.datetime.fromtimestamp(utc, datetime.timezone.utc)} UTC)")
|
||
|
|
|
||
|
|
if not args.apply:
|
||
|
|
print(f"\nDry run: would rewrite {seen} rows. Re-run with --apply.")
|
||
|
|
conn.close()
|
||
|
|
return 0
|
||
|
|
|
||
|
|
if written != to_process:
|
||
|
|
print(f"ERROR: wrote {written} rows but expected {to_process}.")
|
||
|
|
print(" The marker is NOT set; investigate before re-running.")
|
||
|
|
conn.close()
|
||
|
|
return 1
|
||
|
|
|
||
|
|
db.set_meta(conn, MARKER, "done")
|
||
|
|
conn.execute("DELETE FROM meta WHERE key = ?", (CURSOR,))
|
||
|
|
conn.commit()
|
||
|
|
conn.close()
|
||
|
|
print(f"\nRewrote {written} rows and set {MARKER}=done.")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|