~launchpad-pqm/launchpad/devel

# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

"""Database garbage collection."""

__metaclass__ = type

__all__ = [

    'DailyDatabaseGarbageCollector',

    'HourlyDatabaseGarbageCollector',

    ]

from datetime import (

    datetime,

    timedelta,

    )

import logging

import multiprocessing

import os

import threading

import time

from contrib.glock import (

    GlobalLock,

    LockAlreadyAcquired,

    )

from psycopg2 import IntegrityError

import pytz

from storm.expr import In

from storm.locals import (

    Max,

    Min,

    SQL,

    )

import transaction

from zope.component import getUtility

from zope.security.proxy import removeSecurityProxy

from canonical.config import config

from canonical.database import postgresql

from canonical.database.constants import UTC_NOW

from canonical.database.sqlbase import (

    cursor,

    session_store,

    sqlvalues,

    )

from canonical.launchpad.database.emailaddress import EmailAddress

from canonical.launchpad.database.librarian import TimeLimitedToken

from canonical.launchpad.database.logintoken import LoginToken

from canonical.launchpad.database.oauth import OAuthNonce

from canonical.launchpad.database.openidconsumer import OpenIDConsumerNonce

from canonical.launchpad.interfaces.account import AccountStatus

from canonical.launchpad.interfaces.emailaddress import EmailAddressStatus

from canonical.launchpad.interfaces.lpstorm import IMasterStore

from canonical.launchpad.utilities.looptuner import TunableLoop

from canonical.launchpad.webapp.interfaces import (

    IStoreSelector,

    MAIN_STORE,

    MASTER_FLAVOR,

    )

from lp.answers.model.answercontact import AnswerContact

from lp.bugs.interfaces.bug import IBugSet

from lp.bugs.model.bug import Bug

from lp.bugs.model.bugattachment import BugAttachment

from lp.bugs.model.bugnotification import BugNotification

from lp.bugs.model.bugwatch import BugWatchActivity

from lp.bugs.scripts.checkwatches.scheduler import (

    BugWatchScheduler,

    MAX_SAMPLE_SIZE,

    )

from lp.code.interfaces.revision import IRevisionSet

from lp.code.model.codeimportevent import CodeImportEvent

from lp.code.model.codeimportresult import CodeImportResult

from lp.code.model.revision import (

    RevisionAuthor,

    RevisionCache,

    )

from lp.hardwaredb.model.hwdb import HWSubmission

from lp.registry.model.person import Person

from lp.services.job.model.job import Job

from lp.services.log.logger import PrefixFilter

from lp.services.propertycache import cachedproperty

from lp.services.scripts.base import (

    LaunchpadCronScript,

    LOCK_PATH,

    SilentLaunchpadScriptFailure,

    )

from lp.services.session.model import SessionData

from lp.soyuz.model.publishing import (

    BinaryPackagePublishingHistory,

    SourcePackagePublishingHistory,

    )

from lp.translations.interfaces.potemplate import IPOTemplateSet

from lp.translations.model.potmsgset import POTMsgSet

from lp.translations.model.potranslation import POTranslation

from lp.translations.model.translationmessage import TranslationMessage

from lp.translations.model.translationtemplateitem import (

    TranslationTemplateItem,

    )

ONE_DAY_IN_SECONDS = 24 * 60 * 60

class BulkPruner(TunableLoop):

    """A abstract ITunableLoop base class for simple pruners.

    This is designed for the case where calculating the list of items

    is expensive, and this list may be huge. For this use case, it

    is impractical to calculate a batch of ids to remove each

    iteration.

    One approach is using a temporary table, populating it

    with the set of items to remove at the start. However, this

    approach can perform badly as you either need to prune the

    temporary table as you go, or using OFFSET to skip to the next

    batch to remove which gets slower as we progress further through

    the list.

    Instead, this implementation declares a CURSOR that can be used

    across multiple transactions, allowing us to calculate the set

    of items to remove just once and iterate over it, avoiding the

    seek-to-batch issues with a temporary table and OFFSET yet

    deleting batches of rows in separate transactions.

    """

    # The Storm database class for the table we are removing records

    # from. Must be overridden.

    target_table_class = None

    # The column name in target_table we use as the key. The type must

    # match that returned by the ids_to_prune_query and the

    # target_table_key_type. May be overridden.

    target_table_key = 'id'

    # SQL type of the target_table_key. May be overridden.

    target_table_key_type = 'id integer'

    # An SQL query returning a list of ids to remove from target_table.

    # The query must return a single column named 'id' and should not

    # contain duplicates. Must be overridden.

    ids_to_prune_query = None

    # See `TunableLoop`. May be overridden.

    maximum_chunk_size = 10000

    def getStore(self):

        """The master Store for the table we are pruning.

        May be overridden.

        """

        return IMasterStore(self.target_table_class)

    _unique_counter = 0

    def __init__(self, log, abort_time=None):

        super(BulkPruner, self).__init__(log, abort_time)

        self.store = self.getStore()

        self.target_table_name = self.target_table_class.__storm_table__

        self._unique_counter += 1

        self.cursor_name = (

            'bulkprunerid_%s_%d'

            % (self.__class__.__name__, self._unique_counter)).lower()

        # Open the cursor.

        self.store.execute(

            "DECLARE %s NO SCROLL CURSOR WITH HOLD FOR %s"

            % (self.cursor_name, self.ids_to_prune_query))

    _num_removed = None

    def isDone(self):

        """See `ITunableLoop`."""

        return self._num_removed == 0

    def __call__(self, chunk_size):

        """See `ITunableLoop`."""

        result = self.store.execute("""

            DELETE FROM %s

            WHERE (%s) IN (

                SELECT * FROM

                cursor_fetch('%s', %d) AS f(%s))

            """

            % (

                self.target_table_name, self.target_table_key,

                self.cursor_name, chunk_size, self.target_table_key_type))

        self._num_removed = result.rowcount

        transaction.commit()

    def cleanUp(self):

        """See `ITunableLoop`."""

        self.store.execute("CLOSE %s" % self.cursor_name)

class LoginTokenPruner(BulkPruner):

    """Remove old LoginToken rows.

    After 1 year, they are useless even for archaeology.

    """

    target_table_class = LoginToken

    ids_to_prune_query = """

        SELECT id FROM LoginToken WHERE

        created < CURRENT_TIMESTAMP - CAST('1 year' AS interval)

        """

class POTranslationPruner(BulkPruner):

    """Remove unlinked POTranslation entries.

    XXX bug=723596 StuartBishop: This job only needs to run once per month.

    """

    target_table_class = POTranslation

    ids_to_prune_query = """

        SELECT POTranslation.id AS id FROM POTranslation

        EXCEPT (

            SELECT msgstr0 FROM TranslationMessage

                WHERE msgstr0 IS NOT NULL

            UNION ALL SELECT msgstr1 FROM TranslationMessage

                WHERE msgstr1 IS NOT NULL

            UNION ALL SELECT msgstr2 FROM TranslationMessage

                WHERE msgstr2 IS NOT NULL

            UNION ALL SELECT msgstr3 FROM TranslationMessage

                WHERE msgstr3 IS NOT NULL

            UNION ALL SELECT msgstr4 FROM TranslationMessage

                WHERE msgstr4 IS NOT NULL

            UNION ALL SELECT msgstr5 FROM TranslationMessage

                WHERE msgstr5 IS NOT NULL

            )

        """

class SessionPruner(BulkPruner):

    """Base class for session removal."""

    target_table_class = SessionData

    target_table_key = 'client_id'

    target_table_key_type = 'id text'

class AntiqueSessionPruner(SessionPruner):

    """Remove sessions not accessed for 60 days"""

    ids_to_prune_query = """

        SELECT client_id AS id FROM SessionData

        WHERE last_accessed < CURRENT_TIMESTAMP - CAST('60 days' AS interval)

        """

class UnusedSessionPruner(SessionPruner):

    """Remove sessions older than 1 day with no authentication credentials."""

    ids_to_prune_query = """

        SELECT client_id AS id FROM SessionData

        WHERE

            last_accessed < CURRENT_TIMESTAMP - CAST('1 day' AS interval)

            AND client_id NOT IN (

                SELECT client_id

                FROM SessionPkgData

                WHERE

                    product_id = 'launchpad.authenticateduser'

                    AND key='logintime')

        """

class DuplicateSessionPruner(SessionPruner):

    """Remove all but the most recent 6 authenticated sessions for a user.

    We sometimes see users with dozens or thousands of authenticated

    sessions. To limit exposure to replay attacks, we remove all but

    the most recent 6 of them for a given user.

    """

    ids_to_prune_query = """

        SELECT client_id AS id

        FROM (

            SELECT

                sessiondata.client_id,

                last_accessed,

                rank() OVER pickle AS rank

            FROM SessionData, SessionPkgData

            WHERE

                SessionData.client_id = SessionPkgData.client_id

                AND product_id = 'launchpad.authenticateduser'

                AND key='accountid'

            WINDOW pickle AS (PARTITION BY pickle ORDER BY last_accessed DESC)

            ) AS whatever

        WHERE

            rank > 6

            AND last_accessed < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'

                - CAST('1 hour' AS interval)

        """

class OAuthNoncePruner(BulkPruner):

    """An ITunableLoop to prune old OAuthNonce records.

    We remove all OAuthNonce records older than 1 day.

    """

    target_table_key = 'access_token, request_timestamp, nonce'

    target_table_key_type = (

        'access_token integer, request_timestamp timestamp without time zone,'

        ' nonce text')

    target_table_class = OAuthNonce

    ids_to_prune_query = """

        SELECT access_token, request_timestamp, nonce FROM OAuthNonce

        WHERE request_timestamp

            < CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - CAST('1 day' AS interval)

        """

class BugSummaryJournalRollup(TunableLoop):

    """Rollup BugSummaryJournal rows into BugSummary."""

    maximum_chunk_size = 5000

    def __init__(self, log, abort_time=None):

        super(BugSummaryJournalRollup, self).__init__(log, abort_time)

        self.store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)

    def isDone(self):

        has_more = self.store.execute(

            "SELECT EXISTS (SELECT TRUE FROM BugSummaryJournal LIMIT 1)"

            ).get_one()[0]

        return not has_more

    def __call__(self, chunk_size):

        chunk_size = int(chunk_size + 0.5)

        self.store.execute(

            "SELECT bugsummary_rollup_journal(%s)", (chunk_size,),

            noresult=True)

        self.store.commit()

class OpenIDConsumerNoncePruner(TunableLoop):

    """An ITunableLoop to prune old OpenIDConsumerNonce records.

    We remove all OpenIDConsumerNonce records older than 1 day.

    """

    maximum_chunk_size = 6 * 60 * 60  # 6 hours in seconds.

    def __init__(self, log, abort_time=None):

        super(OpenIDConsumerNoncePruner, self).__init__(log, abort_time)

        self.store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)

        self.earliest_timestamp = self.store.find(

            Min(OpenIDConsumerNonce.timestamp)).one()

        utc_now = int(time.mktime(time.gmtime()))

        self.earliest_wanted_timestamp = utc_now - ONE_DAY_IN_SECONDS

    def isDone(self):

        return (

            self.earliest_timestamp is None

            or self.earliest_timestamp >= self.earliest_wanted_timestamp)

    def __call__(self, chunk_size):

        self.earliest_timestamp = min(

            self.earliest_wanted_timestamp,

            self.earliest_timestamp + chunk_size)

        self.log.debug(

            "Removing OpenIDConsumerNonce rows older than %s"

            % self.earliest_timestamp)

        self.store.find(

            OpenIDConsumerNonce,

            OpenIDConsumerNonce.timestamp < self.earliest_timestamp).remove()

        transaction.commit()

class OpenIDConsumerAssociationPruner(TunableLoop):

    minimum_chunk_size = 3500

    maximum_chunk_size = 50000

    table_name = 'OpenIDConsumerAssociation'

    _num_removed = None

    def __init__(self, log, abort_time=None):

        super(OpenIDConsumerAssociationPruner, self).__init__(log, abort_time)

        self.store = getUtility(IStoreSelector).get(MAIN_STORE, MASTER_FLAVOR)

    def __call__(self, chunksize):

        result = self.store.execute("""

            DELETE FROM %s

            WHERE (server_url, handle) IN (

                SELECT server_url, handle FROM %s

                WHERE issued + lifetime <

                    EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)

                LIMIT %d

                )

            """ % (self.table_name, self.table_name, int(chunksize)))

        self._num_removed = result.rowcount

        transaction.commit()

    def isDone(self):

        return self._num_removed == 0

class RevisionCachePruner(TunableLoop):

    """A tunable loop to remove old revisions from the cache."""

    maximum_chunk_size = 100

    def isDone(self):

        """We are done when there are no old revisions to delete."""

        epoch = datetime.now(pytz.UTC) - timedelta(days=30)

        store = IMasterStore(RevisionCache)

        results = store.find(

            RevisionCache, RevisionCache.revision_date < epoch)

        return results.count() == 0

    def __call__(self, chunk_size):

        """Delegate to the `IRevisionSet` implementation."""

        getUtility(IRevisionSet).pruneRevisionCache(chunk_size)

        transaction.commit()

class CodeImportEventPruner(BulkPruner):

    """Prune `CodeImportEvent`s that are more than a month old.

    Events that happened more than 30 days ago are really of no

    interest to us.

    """

    target_table_class = CodeImportEvent

    ids_to_prune_query = """

        SELECT id FROM CodeImportEvent

        WHERE date_created < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'

            - CAST('30 days' AS interval)

        """

class CodeImportResultPruner(BulkPruner):

    """A TunableLoop to prune unwanted CodeImportResult rows.

    Removes CodeImportResult rows if they are older than 30 days

    and they are not one of the most recent results for that

    CodeImport.

    """

    target_table_class = CodeImportResult

    ids_to_prune_query = """

        SELECT id FROM (

            SELECT id, date_created, rank() OVER w AS rank

            FROM CodeImportResult

            WINDOW w AS (PARTITION BY code_import ORDER BY date_created DESC)

            ) AS whatever

        WHERE

            rank > %s

            AND date_created < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'

                - CAST('30 days' AS interval)

            """ % sqlvalues(config.codeimport.consecutive_failure_limit - 1)

class RevisionAuthorEmailLinker(TunableLoop):

    """A TunableLoop that links `RevisionAuthor` objects to `Person` objects.

    `EmailAddress` objects are looked up for `RevisionAuthor` objects

    that have not yet been linked to a `Person`.  If the

    `EmailAddress` is linked to a person, then the `RevisionAuthor` is

    linked to the same.

    """

    maximum_chunk_size = 1000

    def __init__(self, log, abort_time=None):

        super(RevisionAuthorEmailLinker, self).__init__(log, abort_time)

        self.author_store = IMasterStore(RevisionAuthor)

        self.email_store = IMasterStore(EmailAddress)

        (self.min_author_id,

         self.max_author_id) = self.author_store.find(

            (Min(RevisionAuthor.id), Max(RevisionAuthor.id))).one()

        self.next_author_id = self.min_author_id

    def isDone(self):

        return (self.min_author_id is None or

                self.next_author_id > self.max_author_id)

    def __call__(self, chunk_size):

        result = self.author_store.find(

            RevisionAuthor,

            RevisionAuthor.id >= self.next_author_id,

            RevisionAuthor.personID == None,

            RevisionAuthor.email != None)

        result.order_by(RevisionAuthor.id)

        authors = list(result[:chunk_size])

        # No more authors found.

        if len(authors) == 0:

            self.next_author_id = self.max_author_id + 1

            transaction.commit()

            return

        emails = dict(self.email_store.find(

            (EmailAddress.email.lower(), EmailAddress.personID),

            EmailAddress.email.lower().is_in(

                    [author.email.lower() for author in authors]),

            EmailAddress.status.is_in([EmailAddressStatus.PREFERRED,

                                       EmailAddressStatus.VALIDATED]),

            EmailAddress.personID != None))

        if emails:

            for author in authors:

                personID = emails.get(author.email.lower())

                if personID is None:

                    continue

                author.personID = personID

        self.next_author_id = authors[-1].id + 1

        transaction.commit()

class HWSubmissionEmailLinker(TunableLoop):

    """A TunableLoop that links `HWSubmission` objects to `Person` objects.

    `EmailAddress` objects are looked up for `HWSubmission` objects

    that have not yet been linked to a `Person`.  If the

    `EmailAddress` is linked to a person, then the `HWSubmission` is

    linked to the same.

    """

    maximum_chunk_size = 50000

    def __init__(self, log, abort_time=None):

        super(HWSubmissionEmailLinker, self).__init__(log, abort_time)

        self.submission_store = IMasterStore(HWSubmission)

        self.submission_store.execute(

            "DROP TABLE IF EXISTS NewlyMatchedSubmission")

        # The join with the Person table is to avoid any replication

        # lag issues - EmailAddress.person might reference a Person

        # that does not yet exist.

        self.submission_store.execute("""

            CREATE TEMPORARY TABLE NewlyMatchedSubmission AS

            SELECT

                HWSubmission.id AS submission,

                EmailAddress.person AS owner

            FROM HWSubmission, EmailAddress, Person

            WHERE HWSubmission.owner IS NULL

                AND EmailAddress.person = Person.id

                AND EmailAddress.status IN %s

                AND lower(HWSubmission.raw_emailaddress)

                    = lower(EmailAddress.email)

            """ % sqlvalues(

                [EmailAddressStatus.VALIDATED, EmailAddressStatus.PREFERRED]),

            noresult=True)

        self.submission_store.execute("""

            CREATE INDEX newlymatchsubmission__submission__idx

            ON NewlyMatchedSubmission(submission)

            """, noresult=True)

        self.matched_submission_count = self.submission_store.execute("""

            SELECT COUNT(*) FROM NewlyMatchedSubmission

            """).get_one()[0]

        self.offset = 0

    def isDone(self):

        return self.offset >= self.matched_submission_count

    def __call__(self, chunk_size):

        self.submission_store.execute("""

            UPDATE HWSubmission

            SET owner=NewlyMatchedSubmission.owner

            FROM (

                SELECT submission, owner

                FROM NewlyMatchedSubmission

                ORDER BY submission

                OFFSET %d

                LIMIT %d

                ) AS NewlyMatchedSubmission

            WHERE HWSubmission.id = NewlyMatchedSubmission.submission

            """ % (self.offset, chunk_size), noresult=True)

        self.offset += chunk_size

        transaction.commit()

class PersonPruner(TunableLoop):

    maximum_chunk_size = 1000

    def __init__(self, log, abort_time=None):

        super(PersonPruner, self).__init__(log, abort_time)

        self.offset = 1

        self.store = IMasterStore(Person)

        self.log.debug("Creating LinkedPeople temporary table.")

        self.store.execute(

            "CREATE TEMPORARY TABLE LinkedPeople(person integer primary key)")

        # Prefill with Person entries created after our OpenID provider

        # started creating personless accounts on signup.

        self.log.debug(

            "Populating LinkedPeople with post-OpenID created Person.")

        self.store.execute("""

            INSERT INTO LinkedPeople

            SELECT id FROM Person

            WHERE datecreated > '2009-04-01'

            """)

        transaction.commit()

        for (from_table, from_column, to_table, to_column, uflag, dflag) in (

                postgresql.listReferences(cursor(), 'person', 'id')):

            # Skip things that don't link to Person.id or that link to it from

            # TeamParticipation or EmailAddress, as all Person entries will be

            # linked to from these tables.  Similarly, PersonSettings can

            # simply be deleted if it exists, because it has a 1 (or 0) to 1

            # relationship with Person.

            if (to_table != 'person' or to_column != 'id'

                or from_table in ('teamparticipation', 'emailaddress',

                                  'personsettings')):

                continue

            self.log.debug(

                "Populating LinkedPeople from %s.%s"

                % (from_table, from_column))

            self.store.execute("""

                INSERT INTO LinkedPeople

                SELECT DISTINCT %(from_column)s AS person

                FROM %(from_table)s

                WHERE %(from_column)s IS NOT NULL

                EXCEPT ALL

                SELECT person FROM LinkedPeople

                """ % dict(from_table=from_table, from_column=from_column))

            transaction.commit()

        self.log.debug("Creating UnlinkedPeople temporary table.")

        self.store.execute("""

            CREATE TEMPORARY TABLE UnlinkedPeople(

                id serial primary key, person integer);

            """)

        self.log.debug("Populating UnlinkedPeople.")

        self.store.execute("""

            INSERT INTO UnlinkedPeople (person) (

                SELECT id AS person FROM Person

                WHERE teamowner IS NULL

                EXCEPT ALL

                SELECT person FROM LinkedPeople);

            """)

        transaction.commit()

        self.log.debug("Indexing UnlinkedPeople.")

        self.store.execute("""

            CREATE UNIQUE INDEX unlinkedpeople__person__idx ON

                UnlinkedPeople(person);

            """)

        self.log.debug("Analyzing UnlinkedPeople.")

        self.store.execute("""

            ANALYZE UnlinkedPeople;

            """)

        self.log.debug("Counting UnlinkedPeople.")

        self.max_offset = self.store.execute(

            "SELECT MAX(id) FROM UnlinkedPeople").get_one()[0]

        if self.max_offset is None:

            self.max_offset = -1  # Trigger isDone() now.

            self.log.debug("No Person records to remove.")

        else:

            self.log.info("%d Person records to remove." % self.max_offset)

        # Don't keep any locks open - we might block.

        transaction.commit()

    def isDone(self):

        return self.offset > self.max_offset

    def __call__(self, chunk_size):

        subquery = """

            SELECT person FROM UnlinkedPeople

            WHERE id BETWEEN %d AND %d

            """ % (self.offset, self.offset + chunk_size - 1)

        people_ids = ",".join(

            str(item[0]) for item in self.store.execute(subquery).get_all())

        self.offset += chunk_size

        try:

            # This would be dangerous if we were deleting a

            # team, so join with Person to ensure it isn't one

            # even in the rare case a person is converted to

            # a team during this run.

            self.store.execute("""

                DELETE FROM TeamParticipation

                USING Person

                WHERE TeamParticipation.person = Person.id

                    AND Person.teamowner IS NULL

                    AND Person.id IN (%s)

                """ % people_ids)

            self.store.execute("""

                UPDATE EmailAddress SET person=NULL

                WHERE person IN (%s)

                """ % people_ids)

            # This cascade deletes any PersonSettings records.

            self.store.execute("""

                DELETE FROM Person

                WHERE id IN (%s)

                """ % people_ids)

            transaction.commit()

            self.log.debug(

                "Deleted the following unlinked people: %s" % people_ids)

        except IntegrityError:

            # This case happens when a Person is linked to something

            # during the run. It is unlikely to occur, so just ignore

            # it again. Everything will clear up next run.

            transaction.abort()

            self.log.warning(

                "Failed to delete %d Person records. Left for next time."

                % chunk_size)

class BugNotificationPruner(BulkPruner):

    """Prune `BugNotificationRecipient` records no longer of interest.

    We discard all rows older than 30 days that have been sent. We

    keep 30 days worth or records to help diagnose email delivery issues.

    """

    target_table_class = BugNotification

    ids_to_prune_query = """

        SELECT BugNotification.id FROM BugNotification

        WHERE date_emailed < CURRENT_TIMESTAMP AT TIME ZONE 'UTC'

            - CAST('30 days' AS interval)

        """

class AnswerContactPruner(BulkPruner):

    """Remove old answer contacts which are no longer required.

    Remove a person as an answer contact if:

      their account has been deactivated for more than one day, or

      suspended for more than one week.

    """

    target_table_class = AnswerContact

    ids_to_prune_query = """

        SELECT DISTINCT AnswerContact.id

        FROM AnswerContact, Person, Account

        WHERE

            AnswerContact.person = Person.id

            AND Person.account = Account.id

            AND (

                (Account.date_status_set <

                CURRENT_TIMESTAMP AT TIME ZONE 'UTC'

                - CAST('1 day' AS interval)

                AND Account.status = %s)

                OR

                (Account.date_status_set <

                CURRENT_TIMESTAMP AT TIME ZONE 'UTC'

                - CAST('7 days' AS interval)

                AND Account.status = %s)

            )

        """ % (AccountStatus.DEACTIVATED.value, AccountStatus.SUSPENDED.value)