~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

"""Librarian garbage collection routines"""

__metaclass__ = type

from datetime import datetime, timedelta
import errno
import re
import sys
from time import time
import os

from zope.interface import implements

from canonical.config import config
from canonical.database.postgresql import drop_tables, quoteIdentifier
from lp.services.looptuner import (
    DBLoopTuner,
    ITunableLoop,
    )
from canonical.librarian.storage import _relFileLocation as relative_file_path
from canonical.librarian.storage import _sameFile
from canonical.database.postgresql import listReferences

log = None  # This is set by cronscripts/librarian-gc.py
debug = False


def confirm_no_clock_skew(con):
    """Raise an exception if there is significant clock skew between the
    database and this machine.

    It is theoretically possible to lose data if there is more than several
    hours of skew.
    """
    cur = con.cursor()
    cur.execute("SELECT CURRENT_TIMESTAMP AT TIME ZONE 'UTC'")
    db_now = cur.fetchone()[0]
    local_now = datetime.utcnow()
    five_minutes = timedelta(minutes=5)

    if -five_minutes < local_now - db_now < five_minutes:
        return
    else:
        raise Exception("%s clock skew between librarian and database" % (
            local_now - db_now,
            ))


def delete_expired_blobs(con):
    """Remove expired TemporaryBlobStorage entries and their corresponding
       LibraryFileAlias entries.

       We delete the LibraryFileAliases here as the default behavior of the
       garbage collector could leave them hanging around indefinitely.

       We also delete any linked ApportJob and Job records here.
    """
    cur = con.cursor()

    # Generate the list of expired blobs.
    cur.execute("""
        SELECT file_alias
        INTO TEMPORARY TABLE BlobAliasesToDelete
        FROM LibraryFileAlias, TemporaryBlobStorage
        WHERE file_alias = LibraryFileAlias.id
            AND expires < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
        """)

    # Generate the list of expired Jobs. We ignore jobs that have not
    # finished.
    cur.execute("""
        SELECT job
        INTO TEMPORARY TABLE JobsToDelete
        FROM Job, ApportJob, TemporaryBlobStorage, LibraryFileAlias
        WHERE
            ApportJob.blob = TemporaryBlobStorage.id
            AND Job.id = ApportJob.job
            AND Job.date_finished < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
            AND TemporaryBlobStorage.file_alias = LibraryFileAlias.id
                AND expires < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
        """)

    # Delete expired ApportJob records.
    cur.execute("""
        DELETE FROM ApportJob
        USING JobsToDelete
        WHERE ApportJob.job = JobsToDelete.job
        """)

    # Delete expired Job records.
    cur.execute("""
        DELETE FROM Job
        USING JobsToDelete
        WHERE Job.id = JobsToDelete.job
        """)

    # Delete expired blobs.
    cur.execute("""
        DELETE FROM TemporaryBlobStorage
        USING BlobAliasesToDelete
        WHERE TemporaryBlobStorage.file_alias = BlobAliasesToDelete.file_alias
        """)

    # Delete LibraryFileAliases referencing expired blobs.
    cur.execute("""
        DELETE FROM LibraryFileAlias
        USING BlobAliasesToDelete
        WHERE file_alias = LibraryFileAlias.id
        """)
    log.info("Removed %d expired blobs" % cur.rowcount)
    con.commit()


def merge_duplicates(con):
    """Merge duplicate LibraryFileContent rows

    This is the first step in a full garbage collection run. We assume files
    are identical if their sha1 hashes and filesizes are identical. For every
    duplicate detected, we make all LibraryFileAlias entries point to one of
    them and delete the unnecessary duplicates from the filesystem and the
    database.
    """

    # Get a list of all (sha1, filesize) that are duplicated in
    # LibraryFileContent
    cur = con.cursor()
    cur.execute("""
        SELECT sha1, filesize
        FROM LibraryFileContent
        GROUP BY sha1, filesize
        HAVING COUNT(*) > 1
        """)
    rows = list(cur.fetchall())

    # Merge the duplicate entries, each one in a seperate transaction
    for sha1, filesize in rows:
        cur = con.cursor()

        sha1 = sha1.encode('US-ASCII')  # Can't pass Unicode to execute (yet)

        # Get a list of our dupes. Where multiple files exist, we return
        # the most recently added one first, because this is the version
        # most likely to exist on the staging server (it should be
        # irrelevant on production).
        cur.execute("""
            SELECT id
            FROM LibraryFileContent
            WHERE sha1=%(sha1)s AND filesize=%(filesize)s
            ORDER BY datecreated DESC
            """, vars())
        dupes = [row[0] for row in cur.fetchall()]

        if debug:
            log.debug("Found duplicate LibraryFileContents")
            # Spit out more info in case it helps work out where
            # dupes are coming from.
            for dupe_id in dupes:
                cur.execute("""
                    SELECT id, filename, mimetype FROM LibraryFileAlias
                    WHERE content = %(dupe_id)s
                    """, vars())
                for id, filename, mimetype in cur.fetchall():
                    log.debug("> %d %s %s" % (id, filename, mimetype))

        # Make sure the first file exists on disk. Don't merge if it
        # doesn't. This shouldn't happen on production, so we don't try
        # and cope - just report and skip. However, on staging this will
        # be more common because database records has been synced from
        # production but the actual librarian contents has not.
        dupe1_id = dupes[0]
        dupe1_path = get_file_path(dupe1_id)
        if not os.path.exists(dupe1_path):
            if config.instance_name == 'staging':
                log.debug(
                        "LibraryFileContent %d data is missing (%s)",
                        dupe1_id, dupe1_path
                        )
            else:
                log.warning(
                        "LibraryFileContent %d data is missing (%s)",
                        dupe1_id, dupe1_path
                        )
            continue

        # Do a manual check that they really are identical, because we
        # employ paranoids. And we might as well cope with someone breaking
        # SHA1 enough that it becomes possible to create a SHA1 collision
        # with an identical filesize to an existing file. Which is pretty
        # unlikely. Where did I leave my tin foil hat?
        for dupe2_id in (dupe for dupe in dupes[1:]):
            dupe2_path = get_file_path(dupe2_id)
            # Check paths exist, because on staging they may not!
            if (os.path.exists(dupe2_path)
                and not _sameFile(dupe1_path, dupe2_path)):
                log.error(
                        "SHA-1 collision found. LibraryFileContent %d and "
                        "%d have the same SHA1 and filesize, but are not "
                        "byte-for-byte identical.",
                        dupe1_id, dupe2_id
                        )
                sys.exit(1)

        # Update all the LibraryFileAlias entries to point to a single
        # LibraryFileContent
        prime_id = dupes[0]
        other_ids = ', '.join(str(dupe) for dupe in dupes[1:])
        log.debug(
            "Making LibraryFileAliases referencing %s reference %s instead",
            other_ids, prime_id
            )
        for other_id in dupes[1:]:
            cur.execute("""
                UPDATE LibraryFileAlias SET content=%(prime_id)s
                WHERE content = %(other_id)s
                """, vars())

        log.debug("Committing")
        con.commit()


class ExpireAliases:
    """Expire expired LibraryFileAlias records.

    This simply involves setting the LibraryFileAlias.content to NULL.
    Unreferenced LibraryFileContent records are cleaned up elsewhere.
    """
    implements(ITunableLoop)

    def __init__(self, con):
        self.con = con
        self.total_expired = 0
        self._done = False

    def isDone(self):
        if self._done:
            log.info(
                "Expired %d LibraryFileAlias records." % self.total_expired)
            return True
        else:
            return False

    def __call__(self, chunksize):
        chunksize = int(chunksize)
        cur = self.con.cursor()
        cur.execute("""
            UPDATE LibraryFileAlias
            SET content=NULL
            WHERE id IN (
                SELECT id FROM LibraryFileAlias
                WHERE
                    content IS NOT NULL
                    AND expires < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
                        - interval '1 week'
                LIMIT %d)
            """ % chunksize)
        self.total_expired += cur.rowcount
        if cur.rowcount == 0:
            self._done = True
        else:
            log.debug("Expired %d LibraryFileAlias records." % cur.rowcount)
        self.con.commit()


def expire_aliases(con):
    """Invoke ExpireLibraryFileAliases."""
    loop_tuner = DBLoopTuner(ExpireAliases(con), 5, log=log)
    loop_tuner.run()


class UnreferencedLibraryFileAliasPruner:
    """Delete unreferenced LibraryFileAliases.

    The LibraryFileContent records are left untouched for the code that
    knows how to delete them and the corresponding files on disk.

    This is the second step in a full garbage collection sweep. We determine
    which LibraryFileAlias entries are not being referenced by other objects
    in the database and delete them, if they are expired (expiry in the past
    or NULL), and if they have not been recently accessed (last_access over
    one week in the past).
    """
    implements(ITunableLoop)

    def __init__(self, con):
        self.con = con  # Database connection to use
        self.total_deleted = 0  # Running total
        self.index = 1

        log.info("Deleting unreferenced LibraryFileAliases")

        cur = con.cursor()

        drop_tables(cur, "ReferencedLibraryFileAlias")
        cur.execute("""
            CREATE TEMPORARY TABLE ReferencedLibraryFileAlias (
                alias integer)
            """)

        # Determine what columns link to LibraryFileAlias
        # references = [(table, column), ...]
        references = [
            tuple(ref[:2])
            for ref in listReferences(cur, 'libraryfilealias', 'id')
            if ref[0] != 'libraryfiledownloadcount'
            ]
        assert len(references) > 10, (
            'Database introspection returned nonsense')
        log.debug(
            "Found %d columns referencing LibraryFileAlias", len(references))

        # Find all relevant LibraryFileAlias references and fill in
        # ReferencedLibraryFileAlias
        for table, column in references:
            cur.execute("""
                INSERT INTO ReferencedLibraryFileAlias
                SELECT LibraryFileAlias.id
                FROM LibraryFileAlias, %(table)s
                WHERE LibraryFileAlias.id = %(table)s.%(column)s
                """ % {
                    'table': quoteIdentifier(table),
                    'column': quoteIdentifier(column)})
            log.debug("%s.%s references %d LibraryFileContent rows." % (
                table, column, cur.rowcount))
            con.commit()

        log.debug("Calculating unreferenced LibraryFileAlias set.")
        drop_tables(cur, "UnreferencedLibraryFileAlias")
        cur.execute("""
            CREATE TEMPORARY TABLE UnreferencedLibraryFileAlias (
                id serial PRIMARY KEY,
                alias integer UNIQUE)
            """)
        # Calculate the set of unreferenced LibraryFileAlias.
        # We also exclude all unexpired and recently accessed
        # records - we don't remove them even if they are unlinked. We
        # currently don't remove stuff until it has been expired for
        # more than one week, but we will change this if disk space
        # becomes short and it actually will make a noticeable
        # difference. We handle excluding recently created content
        # here rather than earlier when creating the
        # ReferencedLibraryFileAlias table to handle uploads going on
        # while this script is running.
        cur.execute("""
            INSERT INTO UnreferencedLibraryFileAlias (alias)
            SELECT id AS alias FROM LibraryFileAlias
            WHERE
                content IS NULL
                OR ((expires IS NULL OR
                     expires <
                         CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
                             - interval '1 week'
                    )
                    AND last_accessed <
                        CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
                            - interval '1 week'
                    AND date_created <
                        CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
                            - interval '1 week'
                   )
            EXCEPT
            SELECT alias FROM ReferencedLibraryFileAlias
            """)
        con.commit()
        drop_tables(cur, "ReferencedLibraryFileAlias")
        cur.execute(
            "SELECT COALESCE(max(id),0) FROM UnreferencedLibraryFileAlias")
        self.max_id = cur.fetchone()[0]
        log.debug(
            "%d unreferenced LibraryFileContent to remove." % self.max_id)
        con.commit()

    def isDone(self):
        if self.index > self.max_id:
            log.info(
                "Deleted %d LibraryFileAlias records." % self.total_deleted)
            return True
        else:
            return False

    def __call__(self, chunksize):
        chunksize = int(chunksize)
        cur = self.con.cursor()
        cur.execute("""
            DELETE FROM LibraryFileAlias
            WHERE id IN
                (SELECT alias FROM UnreferencedLibraryFileAlias
                WHERE id BETWEEN %s AND %s)
            """, (self.index, self.index + chunksize - 1))
        deleted_rows = cur.rowcount
        self.total_deleted += deleted_rows
        log.debug("Deleted %d LibraryFileAlias records." % deleted_rows)
        self.con.commit()
        self.index += chunksize


def delete_unreferenced_aliases(con):
    "Run the UnreferencedLibraryFileAliasPruner."
    loop_tuner = DBLoopTuner(
        UnreferencedLibraryFileAliasPruner(con), 5, log=log)
    loop_tuner.run()


class UnreferencedContentPruner:
    """Delete LibraryFileContent entries and their disk files that are
    not referenced by any LibraryFileAlias entries.

    Note that a LibraryFileContent can only be accessed through a
    LibraryFileAlias, so all entries in this state are garbage no matter
    what their expires flag says.
    """
    implements(ITunableLoop)

    def __init__(self, con):
        self.con = con
        self.index = 1
        self.total_deleted = 0
        cur = con.cursor()
        drop_tables(cur, "UnreferencedLibraryFileContent")
        cur.execute("""
            CREATE TEMPORARY TABLE UnreferencedLibraryFileContent (
                id serial PRIMARY KEY,
                content integer UNIQUE)
            """)
        cur.execute("""
            INSERT INTO UnreferencedLibraryFileContent (content)
            SELECT DISTINCT LibraryFileContent.id
            FROM LibraryFileContent
            LEFT OUTER JOIN LibraryFileAlias
                ON LibraryFileContent.id = LibraryFileAlias.content
            WHERE LibraryFileAlias.content IS NULL
        """)
        cur.execute("""
            SELECT COALESCE(max(id), 0) FROM UnreferencedLibraryFileContent
            """)
        self.max_id = cur.fetchone()[0]
        log.debug(
            "%d unreferenced LibraryFileContent rows to remove."
            % self.max_id)

    def isDone(self):
        if self.index > self.max_id:
            log.info("Deleted %d unreferenced files." % self.total_deleted)
            return True
        else:
            return False

    def __call__(self, chunksize):
        chunksize = int(chunksize)

        cur = self.con.cursor()

        # Delete unreferenced LibraryFileContent entries.
        cur.execute("""
            DELETE FROM LibraryFileContent
            USING (
                SELECT content FROM UnreferencedLibraryFileContent
                WHERE id BETWEEN %s AND %s) AS UnreferencedLibraryFileContent
            WHERE
                LibraryFileContent.id = UnreferencedLibraryFileContent.content
            """, (self.index, self.index + chunksize - 1))
        rows_deleted = cur.rowcount
        self.total_deleted += rows_deleted
        self.con.commit()

        # Remove files from disk. We do this outside the transaction,
        # as the garbage collector happily deals with files that exist
        # on disk but not in the DB.
        cur.execute("""
            SELECT content FROM UnreferencedLibraryFileContent
            WHERE id BETWEEN %s AND %s
            """, (self.index, self.index + chunksize - 1))
        for content_id in (row[0] for row in cur.fetchall()):
            # Remove the file from disk, if it hasn't already been
            path = get_file_path(content_id)
            try:
                os.unlink(path)
            except OSError, e:
                if e.errno != errno.ENOENT:
                    raise
                if config.librarian_server.upstream_host is None:
                    # It is normal to have files in the database that
                    # are not on disk if the Librarian has an upstream
                    # Librarian, such as on staging. Don't annoy the
                    # operator with noise in this case.
                    log.info("%s already deleted", path)
            else:
                log.debug("Deleted %s", path)
        self.con.rollback()

        self.index += chunksize


def delete_unreferenced_content(con):
    """Invoke UnreferencedContentPruner."""
    loop_tuner = DBLoopTuner(UnreferencedContentPruner(con), 5, log=log)
    loop_tuner.run()


def delete_unwanted_files(con):
    """Delete files found on disk that have no corresponding record in the
    database.

    Files will only be deleted if they were created more than one day ago
    to avoid deleting files that have just been uploaded but have yet to have
    the database records committed.
    """
    cur = con.cursor()

    # Calculate all stored LibraryFileContent ids that we want to keep.
    # Results are ordered so we don't have to suck them all in at once.
    cur.execute("""
        SELECT id FROM LibraryFileContent ORDER BY id
        """)

    def get_next_wanted_content_id():
        result = cur.fetchone()
        if result is None:
            return None
        else:
            return result[0]

    removed_count = 0
    content_id = next_wanted_content_id = -1

    hex_content_id_re = re.compile('^[0-9a-f]{8}$')
    ONE_DAY = 24 * 60 * 60

    for dirpath, dirnames, filenames in os.walk(
        get_storage_root(), followlinks=True):

        # Ignore known and harmless noise in the Librarian storage area.
        if 'incoming' in dirnames:
            dirnames.remove('incoming')
        if 'lost+found' in dirnames:
            dirnames.remove('lost+found')
        filenames = set(filenames)
        filenames.discard('librarian.pid')
        filenames.discard('librarian.log')

        for dirname in dirnames[:]:
            if len(dirname) != 2:
                dirnames.remove(dirname)
                log.warning(
                    "Ignoring directory %s that shouldn't be here" % dirname)
                continue
            try:
                int(dirname, 16)
            except ValueError:
                dirnames.remove(dirname)
                log.warning("Ignoring invalid directory %s" % dirname)

        # We need everything in order to ensure we visit files in the
        # same order we retrieve wanted files from the database.
        dirnames.sort()
        filenames = sorted(filenames)

        # Noise in the storage area, or maybe we are looking at the wrong
        # path?
        if dirnames and filenames:
            log.warning(
                "%s contains both files %r and subdirectories %r. Skipping."
                % (dirpath, filenames, dirnames))
            continue

        for filename in filenames:
            path = os.path.join(dirpath, filename)
            hex_content_id = ''.join(path.split(os.sep)[-4:])
            if hex_content_id_re.search(hex_content_id) is None:
                log.warning(
                    "Ignoring invalid path %s" % path)
                continue

            content_id = int(hex_content_id, 16)

            while (next_wanted_content_id is not None
                    and content_id > next_wanted_content_id):

                next_wanted_content_id = get_next_wanted_content_id()

                if (config.librarian_server.upstream_host is None
                        and next_wanted_content_id is not None
                        and next_wanted_content_id < content_id):
                    log.error(
                        "LibraryFileContent %d exists in the database but "
                        "was not found on disk." % next_wanted_content_id)

            file_wanted = (
                    next_wanted_content_id is not None
                    and next_wanted_content_id == content_id)

            if not file_wanted:
                if time() - os.path.getctime(path) < ONE_DAY:
                    log.debug(
                        "File %d not removed - created too recently"
                        % content_id)
                else:
                    # File uploaded a while ago but no longer wanted.
                    os.unlink(path)
                    log.debug("Deleted %s" % path)
                    removed_count += 1

    # Report any remaining LibraryFileContent that the database says
    # should exist but we didn't find on disk.
    if next_wanted_content_id == content_id:
        next_wanted_content_id = get_next_wanted_content_id()
    while next_wanted_content_id is not None:
        log.error(
            "LibraryFileContent %d exists in the database but "
            "was not found on disk." % next_wanted_content_id)
        next_wanted_content_id = get_next_wanted_content_id()

    log.info(
            "Deleted %d files from disk that where no longer referenced "
            "in the db" % removed_count
            )


def get_file_path(content_id):
    """Return the physical file path to the matching LibraryFileContent id.
    """
    assert isinstance(content_id, (int, long)), (
        'Invalid content_id %s' % repr(content_id))
    return os.path.join(get_storage_root(), relative_file_path(content_id))


def get_storage_root():
    """Return the path to the root of the Librarian storage area.

    Performs some basic sanity checking to avoid accidents.
    """
    storage_root = config.librarian_server.root
    # Do a basic sanity check.
    assert os.path.isdir(os.path.join(storage_root, 'incoming')), (
        '%s is not a Librarian storage area' % storage_root)
    return storage_root