~james-w/udd/storm-sqlite

#!/usr/bin/python

import cgi

import datetime

import errno

import fcntl

import operator

import os

import random

import re

import shutil

import simplejson

import sqlite3

import StringIO

import sys

import time

import urllib

import urllib2

import urlparse

from debian_bundle import changelog

from bzrlib import branch, diff, errors, merge, tag, testament, transport, ui

from bzrlib.trace import mutter

from launchpadlib.credentials import Credentials

from launchpadlib.errors import HTTPError

from launchpadlib.launchpad import Launchpad, EDGE_SERVICE_ROOT

SERVICE_ROOT = EDGE_SERVICE_ROOT.replace("edge.", "")

base_dir = "/srv/package-import.canonical.com/new/"

if not os.path.exists(base_dir):

    base_dir = "."

logs_dir = os.path.join(base_dir, "logs")

retry_dir = os.path.join(logs_dir, "retry")

priority_dir = os.path.join(logs_dir, "priority")

lists_dir = os.path.join(base_dir, "lists")

revids_dir = os.path.join(base_dir, "revids")

lp_creds_file = os.path.join(base_dir, "lp_creds.txt")

last_time_file = os.path.join(base_dir, "last_run")

stop_file = os.path.join(base_dir, "STOP_PLEASE")

debug_log_file = os.path.join(base_dir, "debug_log")

progress_log_file = os.path.join(base_dir, "progress_log")

max_threads_file = os.path.join(base_dir, "max_threads")

locks_dir = os.path.join(base_dir, "locks")

updates_dir = os.path.join(base_dir, "updates")

sqlite_file = os.path.join(base_dir, "meta.db")

sqlite_package_file = os.path.join(base_dir, "packages.db")

sqlite_old_failures_file = os.path.join(base_dir, "failures.db")

sqlite_history_file = os.path.join(base_dir, "history.db")

explanations_file = os.path.join(base_dir, "explanations")

sqlite_timeout = 30

web_base_dir = logs_dir

debian_diffs_dir = os.path.join(web_base_dir, "diffs")

ubuntu_merge_dir = os.path.join(web_base_dir, "merges")

running_sentinel = "Apparently the supervisor died\n"

no_lock_returncode = 139

for directory in (logs_dir, retry_dir, priority_dir, lists_dir,

                  revids_dir, locks_dir, updates_dir,

                  debian_diffs_dir, ubuntu_merge_dir,

                 ):

    if not os.path.exists(directory):

        os.mkdir(directory)

urllib_launchpad_base_url = "https+urllib://launchpad.net/"

urllib_debian_base_url = "http+urllib://ftp.uk.debian.org/debian/"

urllib_debian_archive_base_url = "http+urllib://archive.debian.org/debian/"

distro_releases = {"ubuntu": ["warty", "hoary", "breezy", "dapper", "edgy", "feisty", "gutsy", "hardy", "intrepid", "jaunty", "karmic", "lucid", "maverick"],

    "debian": ["woody", "sarge", "etch", "lenny", "squeeze", "sid", "experimental"],

}

distro_pockets = {"ubuntu": ["release", "security", "proposed", "updates", "backports"],

    "debian": ["release",],

}

# development releases first for stacking, no etch or earlier as they aren't on lp

lp_distro_releases = {"ubuntu": ["maverick", "warty", "hoary", "breezy", "dapper", "edgy", "feisty", "gutsy", "hardy", "intrepid", "jaunty", "karmic", "lucid"],

    "debian": ["squeeze", "lenny", "sid", "experimental"],

}

default_debian_diff_release = "sid"

default_ubuntu_merge_source = "squeeze"

def get_lp(cache=None):

    f = open(lp_creds_file, "rb")

    try:

        creds = Credentials("package-import")

        creds.load(f)

    finally:

        f.close()

    lp = Launchpad(creds, service_root=SERVICE_ROOT, cache=cache)

    return lp

def get_sqlite_connection(path):

    conn = sqlite3.connect(path, timeout=sqlite_timeout,

            detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)

    conn.row_factory = sqlite3.Row

    return conn

def debian_base_url(release):

    if release in ["hamm", "slink", "potato", "woody", "sarge"]:

        return urllib_debian_archive_base_url

    return urllib_debian_base_url

def make_suite(release, pocket):

    if pocket == "release":

        suite = release

    elif pocket in ("security", "proposed", "updates", "backports"):

        suite = "%s-%s" % (release, pocket)

    else:

        assert False, "Unkown pocket: %s" % pocket

    return suite

def _lock_path(path):

    f = open(path, "wb")

    try:

        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)

        return f

    except IOError, e:

        if e.errno in (errno.EACCES, errno.EAGAIN):

            return None

        raise

def lock_package(package):

    path = os.path.join(locks_dir, package)

    return _lock_path(path)

def lock_main():

    path = os.path.join(base_dir, "main_lock")

    return _lock_path(path)

def lock_list_packages():

    path = os.path.join(base_dir, "list_packages")

    return _lock_path(path)

def lock_add_import_jobs():

    path = os.path.join(base_dir, "add_import_jobs")

    return _lock_path(path)

def lock_categorise_failures():

    path = os.path.join(base_dir, "categorise_failures")

    return _lock_path(path)

def lock_email_failures():

    path = os.path.join(base_dir, "email_failures")

    return _lock_path(path)

def lock_update_lists():

    path = os.path.join(base_dir, "update_lists")

    return _lock_path(path)

def lp_call(callable, *args, **kwargs):

    retries = 4

    retried = 0

    while retried < retries:

        try:

            return callable(*args, **kwargs)

        except HTTPError, e:

            retried += 1

            if retried >= retries:

                raise

            if e.response.status // 100 == 5:

                # randomized exponential backoff

                delay = random.randint(10,15)**(retried//2)

                mutter("Sleeping for %d due to %s from LP." % (delay, e))

                time.sleep(delay)

            else:

                if e.response.status == 302:

                    print "Redirected to %s" % e.response.get("location")

                    print "\n".join(["%s: %s" % pair for pair in e.response.items()])

                    print "\n-----\n%s\n----\n" % e.content

                raise

    assert False, "Shouldn't get here"

class PackageToImport(object):

    def __init__(self, name, version, distro, release, pocket, component="",

            on_lp=True, url=None):

        self.name = name

        self.version = version

        self.distro = distro

        self.release = release

        self.component = component

        self.pocket = pocket

        self.on_lp = on_lp

        self.url = url

    def same_as(self, other):

        return (self.name == other.name

                and self.version == other.version

                and self.distro == other.distro

                and self.release == other.release

                and self.pocket == other.pocket)

    def get_url(self):

        if self.url is not None:

            return self.url

        if self.version.debian_version is not None:

            version_str = "%s-%s" % (self.version.upstream_version,

                    self.version.debian_version)

        else:

            version_str = self.version.upstream_version

        if self.distro == "ubuntu":

            files_url = urllib_launchpad_base_url + "ubuntu/%s/+source/%s/%s/+files/" % (self.release, self.name, str(self.version))

        else:

            if self.on_lp:

                files_url = urllib_launchpad_base_url + "debian/%s/+source/%s/%s/+files/" % (self.release, self.name, str(self.version))

            else:

                start = self.name[0]

                if self.name.startswith("lib"):

                    start = self.name[:4]

                files_url = debian_base_url(self.release) + "pool/%s/%s/%s/" % (self.component, start, self.name)

        self.url = files_url + "%s_%s.dsc" % (self.name, version_str)

        return self.url

    def __str__(self):

        return "PackageToImport(%s, %s, %s, %s, %s, %s)" % (self.name,

                self.version, self.distro, self.release, self.pocket,

                self.component)

class ImportList(object):

    def __init__(self):

        self.plist = []

    def sort(self):

        def compare_versions(a, b):

            if a.distro == "ubuntu" and a.distro == b.distro:

                if a.pocket == "backports" and b.pocket != "backports":

                    return -1

                if a.pocket != "backports" and b.pocket == "backports":

                    return 1

            if a.distro != b.distro:

                if a.distro == "debian":

                    return 1

                return -1

            if a.release != b.release:

                a_release = distro_releases[a.distro].index(a.release)

                b_release = distro_releases[b.distro].index(b.release)

                if a_release < b_release:

                    return 1

                return -1

            if a.pocket != b.pocket:

                a_pocket = distro_pockets[a.distro].index(a.pocket)

                b_pocket = distro_pockets[b.distro].index(b.pocket)

                if a_pocket < b_pocket:

                    return 1

                return -1

            if a.version < b.version:

                return 1

            if a.version > b.version:

                return -1

            return 0

        self.plist.sort(cmp=compare_versions)

    def reverse(self):

        self.plist.reverse()

    def add_if_needed(self, newp):

        for oldp in self.plist:

            if oldp.same_as(newp):

                return

        self.plist.append(newp)

    def __str__(self):

        return "[" + ", ".join(map(str, self.plist)) + "]"

def import_from_publications(status_db, publications, last_known_published):

    checked = set()

    newest_published = last_known_published

    for publication in publications:

        published_time = datetime.datetime.strptime(str(publication.date_created)[:19],

                   "%Y-%m-%d %H:%M:%S")

        name = publication.source_package_name

        checked.add(name)

        if (newest_published is None or published_time > newest_published):

            newest_published = published_time

    status_db.add_import_jobs(checked, newest_published)

def create_import_jobs(lp, status_db):

    last_known_published = status_db.last_import_time()

    last_published_call = None

    if last_known_published is not None:

        last_known_published = \

                last_known_published - datetime.timedelta(0, 600)

        last_published_call = last_known_published.isoformat()

    # We want all the new Published records since the last run

    publications = set()

    for p in iterate_collection(lp_call(call_with_limited_size,

            lp.distributions['ubuntu'].main_archive.getPublishedSources,

            status="Published",

            created_since_date=last_published_call)):

        publications.add(p)

    for p in iterate_collection(lp_call(call_with_limited_size,

            lp.distributions['debian'].main_archive.getPublishedSources,

            status="Pending",

            created_since_date=last_published_call)):

        publications.add(p)

    import_from_publications(status_db, publications, last_known_published)

class PackageStatus(object):

    def __init__(self, name):

        self.name = name

        self.failure_count = 0

        self.running = False

        self.queued = False

        self.auto_retry = False

        self.auto_retry_masked = False

        self.timestamp = None

        self.auto_retry_time = None

        self.signature = None

        self.raw_reason = None

class StatusDatabase(object):

    FAILURES_TABLE = "failures"

    FAILURES_TABLE_CREATE = '''create table if not exists %s

                     (package text constraint nonull NOT NULL,

                      reason blob,

                      when_failed timestamp,

                      emailed integer default 0,

                      constraint isprimary PRIMARY KEY

                            (package))''' % FAILURES_TABLE

    FAILURES_TABLE_FIND = '''select * from %s where package=?''' % FAILURES_TABLE

    FAILURES_TABLE_DELETE = '''delete from %s where package=?''' % FAILURES_TABLE

    OLD_FAILURES_TABLE = "old_failures"

    OLD_FAILURES_TABLE_CREATE = '''create table if not exists %s

                     (package text constraint nonull NOT NULL,

                      reason blob,

                      when_failed timestamp,

                      last_failed timestamp,

                      failure_count integer,

                      constraint isprimary PRIMARY KEY

                            (package))''' % OLD_FAILURES_TABLE

    OLD_FAILURES_TABLE_FIND = '''select * from %s where package=?''' % OLD_FAILURES_TABLE

    OLD_FAILURES_TABLE_DELETE = '''delete from %s where package=?''' % OLD_FAILURES_TABLE

    JOBS_TABLE = "jobs"

    JOBS_TABLE_CREATE = '''create table if not exists %s

                    (id integer primary key asc,

                     package text constraint nonull NOT NULL,

                     active integer constraint nonull NOT NULL,

                     type integer constraint nonull NOT NULL,

                     date_requested timestamp constraint nonull NOT NULL,

                     date_started timestamp,

                     date_completed timestamp)''' % JOBS_TABLE

    JOBS_TABLE_FIND = '''select * from %s where package=?

                    and active=? order by type desc''' % JOBS_TABLE

    JOBS_TABLE_NEXT = '''select * from %s where active=1

                    order by type desc''' % JOBS_TABLE

    JOBS_TABLE_SELECT_JOB = '''select * from %s where id=?''' % JOBS_TABLE

    JOBS_TABLE_INSERT = '''insert into %s values (?, ?, ?, ?, ?, ?, ?)''' % JOBS_TABLE

    JOBS_TABLE_UPDATE = '''update %s set package=?, active=?, type=?,

                    date_requested=?, date_started=?, date_completed=?

                    where id=?''' % JOBS_TABLE

    IMPORT_TABLE = "import"

    IMPORT_TABLE_CREATE = '''create table if not exists %s

                    (import timestamp constraint nonull NOT NULL)

                    ''' % IMPORT_TABLE

    RETRY_TABLE = "should_retry"

    RETRY_TABLE_CREATE = '''create table if not exists %s

                    (signature blob constraint nonull NOT NULL,

                     constraint isprimary PRIMARY KEY

                            (signature))''' % RETRY_TABLE

    RETRY_TABLE_SELECT = '''select * from %s where signature=?''' % RETRY_TABLE

    RETRY_TABLE_INSERT = '''insert into %s values (?)''' % RETRY_TABLE

    JOB_TYPE_PRIORITY = 5

    JOB_TYPE_NEW = 3

    JOB_TYPE_RETRY = 1

    AUTO_RETRY_SECONDS = 60*60*3

    MAX_AUTO_RETRY_COUNT = 5

    def __init__(self, sqlite_path):

        self.conn = get_sqlite_connection(sqlite_path)

        c = self.conn.cursor()

        c.execute(self.FAILURES_TABLE_CREATE)

        c.execute(self.OLD_FAILURES_TABLE_CREATE)

        c.execute(self.JOBS_TABLE_CREATE)

        c.execute(self.IMPORT_TABLE_CREATE)

        c.execute(self.RETRY_TABLE_CREATE)

        self.conn.commit()

        c.close()

    def _has_failed(self, c, package):

        rows = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchall()

        return (len(rows) > 0)

    def _set_failure(self, c, package, reason, now):

        if self._has_failed(c, package):

            c.execute('update %s set package=?, reason=?, when_failed=? '

                    'where package=?' % self.FAILURES_TABLE,

                    (package, reason, now, package))

        else:

            c.execute('insert into %s values (?, ?, ?)'

                    % self.FAILURES_TABLE, (package, reason, now))

    def add_jobs_for_interrupted(self):

        c = self.conn.cursor()

        try:

            rows = c.execute('select * from %s where reason=?'

                    % self.FAILURES_TABLE, (running_sentinel,)).fetchall()

            for row in rows:

                c.execute(self.FAILURES_TABLE_DELETE, (row[0],))

                self._add_job(c, row[0], self.JOB_TYPE_PRIORITY)

            self.conn.commit()

        except:

            self.conn.rollback()

            raise

        finally:

            c.close()

    def _start_package(self, c, package):

        ret = None

        rows = c.execute(self.JOBS_TABLE_FIND, (package, 1)).fetchall()

        now = datetime.datetime.utcnow()

        first = now

        job_id = 0

        for row in rows:

            c.execute(self.JOBS_TABLE_UPDATE,

                    (row[1], 0, row[3], row[4], first, row[6], row[0]))

            first = None

            job_id = row[0]

        if self._has_failed(c, package):

            job_id = None

        else:

            c.execute('insert into %s values (?, ?, ?, 0)' % self.FAILURES_TABLE,

                    (package, running_sentinel, now))

            ret = job_id

        return ret

    def start_package(self, package):

        c = self.conn.cursor()

        try:

            ret = self._start_package(c, package)

            self.conn.commit()

            return ret

        except:

            self.conn.rollback()

            raise

        finally:

            c.close()

    def finish_job(self, package, job_id, success, output):

        c = self.conn.cursor()

        try:

            now = datetime.datetime.utcnow()

            if job_id > 0:

                row = c.execute(self.JOBS_TABLE_SELECT_JOB, (job_id,)).fetchone()

                if row is not None:

                    c.execute(self.JOBS_TABLE_UPDATE,

                            (row[1], 0, row[3], row[4], row[5], now, row[0]))

            if success:

                row = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchone()

                if row is not None:

                    c.execute('delete from %s where package=?'

                            % self.JOBS_TABLE, (package,))

                c.execute(self.FAILURES_TABLE_DELETE, (package,))

                c.execute(self.OLD_FAILURES_TABLE_DELETE, (package,))

            else:

                self._set_failure(c, package, output, now)

            self.conn.commit()

        except:

            self.conn.rollback()

            raise

        finally:

            c.close()

    def _add_job(self, c, package, job_type):

        c.execute(self.JOBS_TABLE_INSERT, (None, package, 1, job_type,

                    datetime.datetime.utcnow(), None, None))

    def add_jobs(self, packages, job_type):

        c = self.conn.cursor()

        try:

            for package in packages:

                self._add_job(c, package, job_type)

            self.conn.commit()

        except:

            self.conn.rollback()

            raise

        finally:

            c.close()

    def last_import_time(self):

        c = self.conn.cursor()

        try:

            row = c.execute('select * from %s order by import desc'

                    % self.IMPORT_TABLE).fetchone()

            if row is not None:

                return row[0]

            return row

        finally:

            self.conn.rollback()

            c.close()

    def add_import_jobs(self, packages, newest_published):

        c = self.conn.cursor()

        try:

            for package in packages:

                self._add_job(c, package, self.JOB_TYPE_NEW)

            c.execute('insert into %s values (?)' % self.IMPORT_TABLE,

                    (newest_published,))

            self.conn.commit()

        except:

            self.conn.rollback()

            raise

        finally:

            c.close()

    def next_job(self):

        c = self.conn.cursor()

        try:

            job_id = 0

            package = None

            found = False

            while not found:

                row = c.execute(self.JOBS_TABLE_NEXT).fetchone()

                if row is not None:

                    package = row[1]

                    job_id = self._start_package(c, package)

                    if job_id is None:

                        c.execute(self.JOBS_TABLE_UPDATE,

                                (row[1], 0, row[3], row[4], row[5],

                                 row[6], row[0]))

                    else:

                        found = True

                else:

                    found = True

                    job_id = 0

                    package = None

            self.conn.commit()

            return (job_id, package)

        except:

            self.conn.rollback()

            raise

        finally:

            c.close()

    def failure_reason(self, package):

        c = self.conn.cursor()

        try:

            row = c.execute(self.FAILURES_TABLE_FIND, (package,)).fetchone()

            if row is None:

                return row

            return row[1]

        finally:

            self.conn.rollback()

            c.close()

    def failure_signature(self, raw_reason):

        trace = raw_reason.splitlines()

        sig = ''

        if len(trace) == 1:

            if trace[0] == running_sentinel:

                return None

            # sometimes, Python exceptions do not have file references

            m = re.match('(\w+): ', trace[0])

            if m:

                return m.group(1)

            else:

                return trace[0].strip().replace("\n", " ")

        elif len(trace) < 3:

            return " ".join(trace).strip().replace("\n", " ")

        for l in trace:

            if l.startswith('  File'):

                sig += ':' + l.split()[-1]