[WIP] Split job queueing and job checking

db_migrations/03-create-pendingjobs.sql

@@ -0,0 +1,5 @@
+CREATE TABLE pendingjobs(
+  id INTEGER PRIMARY KEY,
+  job_name TEXT,
+  build_id INTEGER
+);

db_migrations/04-alter-pendingpatches.sql

@@ -0,0 +1,14 @@
+ALTER TABLE pendingpatches RENAME TO pendingpatches_old;
+CREATE TABLE pendingpatches(
+  id INTEGER PRIMARY KEY,
+  patch_id INTEGER UNIQUE,
+  timestamp INTEGER,
+  pendingjob_id INTEGER,
+  FOREIGN KEY(patch_id) REFERENCES patch(id),
+  FOREIGN KEY(pendingjob_id) REFERENCES pendingjobs(id)
+);
+
+INSERT INTO pendingpatches (patch_id, timestamp)
+        SELECT id, timestamp FROM pendingpatches_old;
+
+DROP TABLE pendingpatches_old;

sktm/__init__.py

@@ -179,14 +179,19 @@ def add_pw(self, baseurl, pname, lpatch=None, apikey=None, skip=[]):
 
     # FIXME Fix the name, this function doesn't check anything by itself
     def check_baseline(self):
-        """Submit a build for baseline"""
-        self.pj.append((sktm.jtype.BASELINE,
-                        self.jk.build(self.jobname,
-                                      baserepo=self.baserepo,
-                                      ref=self.baseref,
-                                      baseconfig=self.cfgurl,
-                                      makeopts=self.makeopts),
-                        None))
+        """Submit a build for baseline."""
+        build_id = self.jk.build(
+            self.jobname,
+            baserepo=self.baserepo,
+            ref=self.baseref,
+            baseconfig=self.cfgurl,
+            makeopts=self.makeopts
+        )
+        self.pj.append((sktm.jtype.BASELINE, build_id, None))
+
+        # Add this baseline test to the pendingjobs table
+        self.db.add_pending_job(self.jobname, build_id)
+
 
     def filter_patchsets(self, series_summary_list):
         """
@@ -331,50 +336,66 @@ def check_patchwork(self):
                              series.get_patch_url_list())
 
     def check_pending(self):
-        for (pjt, bid, cpw) in self.pj:
-            if self.jk.is_build_complete(self.jobname, bid):
-                logging.info("job completed: jjid=%d; type=%d", bid, pjt)
-                self.pj.remove((pjt, bid, cpw))
-                if pjt == sktm.jtype.BASELINE:
-                    self.db.update_baseline(
-                        self.baserepo,
-                        self.jk.get_base_hash(self.jobname, bid),
-                        self.jk.get_base_commitdate(self.jobname, bid),
-                        self.jk.get_result(self.jobname, bid),
-                        bid
-                    )
-                elif pjt == sktm.jtype.PATCHWORK:
-                    patches = list()
-                    bres = self.jk.get_result(self.jobname, bid)
-                    rurl = self.jk.get_result_url(self.jobname, bid)
-                    logging.info("result=%s", bres)
-                    logging.info("url=%s", rurl)
-                    basehash = self.jk.get_base_hash(self.jobname, bid)
-                    logging.info("basehash=%s", basehash)
-                    if bres == sktm.tresult.BASELINE_FAILURE:
-                        self.db.update_baseline(
-                            self.baserepo,
-                            basehash,
-                            self.jk.get_base_commitdate(self.jobname, bid),
-                            sktm.tresult.TEST_FAILURE,
-                            bid
-                        )
-
-                    patch_url_list = self.jk.get_patchwork(self.jobname, bid)
-                    for patch_url in patch_url_list:
-                        patches.append(self.get_patch_info_from_url(cpw,
-                                                                    patch_url))
-
-                    if bres != sktm.tresult.BASELINE_FAILURE:
-                        self.db.commit_tested(patches)
-                else:
-                    raise Exception("Unknown job type: %d" % pjt)
-
-    def wait_for_pending(self):
-        self.check_pending()
-        while self.pj:
-            logging.debug("waiting for jobs to complete. %d remaining",
-                          len(self.pj))
-            time.sleep(60)
-            self.check_pending()
-        logging.info("no more pending jobs")
+        """Check on jobs that were sent to Jenkins during the last sktm run."""
+        # Get a list of pending Jenkins jobs
+        pending_jobs = self.db.get_pending_jobs()
+
+        if not pending_jobs:
+            logging.info("No pending jobs to check -- exiting")
+            return
+
+        for pending_job in pending_jobs:
+            pendingjob_id, job_name, build_id = pending_job
+            logging.info("Checking job: %s (#%d)", job_name, build_id)
+
+            # Come back and check on the job later if it is still running.
+            if not self.jk.is_build_complete(job_name, build_id):
+                logging.info(
+                    "Job is still running: %s (#%d)", job_name, build_id
+                )
+                continue
+
+            # Get the build status and check to see if it was aborted.
+            result = self.jk.get_build_status(job_name, build_id)
+            if result == 'ABORTED':
+                logging.info(
+                    "Test was aborted in Jenkins: %s (#%d)",
+                    job_name,
+                    build_id
+                )
+                # Remove the pending job and any associated patches.
+                self.db.remove_pending_job(pendingjob_id)
+                continue
+
+            # Get a list of pending patches associated with this job.
+            pending_patches = self.db.get_patches_for_job(pendingjob_id)
+
+            # Was this a baseline test?
+            if not pending_patches:
+                logging.info(
+                    "Baseline test completed: %s (#%d) [%s]",
+                    job_name,
+                    build_id,
+                    result
+                )
+                # Update the database with the results of the baseline test.
+                self.db.update_baseline(
+                    self.baserepo,
+                    self.jk.get_base_hash(job_name, build_id),
+                    self.jk.get_base_commitdate(job_name, build_id),
+                    self.jk.get_result(job_name, build_id),
+                    build_id
+                )
+                self.db.remove_pending_job(pendingjob_id)
+                return
+
+            # If we made it this far, we are working on a patchwork test.
+            logging.info(
+                "Patchwork test completed: %s (#%d) [%s]",
+                job_name,
+                build_id,
+                result
+            )
+
+            # Clean up the list of pending patches.
+            self.db.commit_tested([x[1] for x in pending_patches])

sktm/db.py

@@ -62,10 +62,17 @@ def __createdb(self, db):
 
                 CREATE TABLE pendingpatches(
                   id INTEGER PRIMARY KEY,
-                  pdate TEXT,
-                  patchsource_id INTEGER,
+                  patch_id INTEGER UNIQUE,
                   timestamp INTEGER,
-                  FOREIGN KEY(patchsource_id) REFERENCES patchsource(id)
+                  pendingjob_id INTEGER,
+                  FOREIGN KEY(patch_id) REFERENCES patch(id),
+                  FOREIGN KEY(pendingjob_id) REFERENCES pendingjobs(id)
+                );
+
+                CREATE TABLE pendingjobs(
+                  id INTEGER PRIMARY KEY,
+                  job_name TEXT,
+                  build_id INTEGER
                 );
 
                 CREATE TABLE testrun(
@@ -154,6 +161,52 @@ def __get_sourceid(self, baseurl, project_id):
 
         return result[0]
 
+    def add_pending_job(self, job_name, build_id):
+        """Add a Jenkins job to the list of pending jobs.
+
+        Args:
+            job_name: Job name in jenkins
+            build_id: Build ID for the Jenkins job
+        """
+        self.cur.execute(
+            "INSERT INTO pendingjobs (job_name, build_id) VALUES (?,?)",
+            (job_name, build_id)
+        )
+        self.conn.commit()
+
+    def get_pending_jobs(self):
+        """Get a list of pending Jenkins jobs."""
+        self.cur.execute("SELECT * FROM pendingjobs")
+        return self.cur.fetchall()
+
+    def remove_pending_job(self, pendingjob_id):
+        """Remove a pending job and any associated pending patches.
+
+        Args:
+            pendingjob_id: ID of a job from the pendingjobs table.
+        """
+        self.cur.execute(
+            "DELETE FROM pendingjobs WHERE id = ?", str(pendingjob_id)
+        )
+        self.cur.execute(
+            "DELETE FROM pendingpatches WHERE pendingjob_id = ?",
+            str(pendingjob_id)
+        )
+        self.conn.commit()
+
+    def get_patches_for_job(self, pendingjob_id):
+        """Get a list of pending patches for a Jenkins job.
+
+        Args:
+            pendingjob_id:  ID of a job from the pendingjobs table.
+        """
+        pendingjob_id = str(pendingjob_id)
+        self.cur.execute(
+            "SELECT * FROM pendingpatches WHERE pendingjob_id = ?",
+            (pendingjob_id)
+        )
+        return self.cur.fetchall()
+
     def get_last_checked_patch(self, baseurl, project_id):
         """Get the patch id of the last patch that was checked.
 
@@ -373,57 +426,36 @@ def __get_latest(self, baserepo):
 
         return result[0]
 
-    def set_patchset_pending(self, baseurl, project_id, series_data):
-        """Add a patch to pendingpatches or update an existing entry.
-
-        Add each specified patch to the list of "pending" patches, with
-        specifed patch date, for specified Patchwork base URL and project ID,
-        and marked with current timestamp. Replace any previously added
-        patches with the same ID (bug: should be "same ID, project ID and
-        base URL").
+    def set_patchset_pending(self, series_data):
+        """Add or update an entry to the pending patches table.
 
         Args:
-            baseurl:     Base URL of the Patchwork instance the project ID and
-                         patch IDs belong to.
-            project_id:  ID of the Patchwork project the patch IDs belong to.
-            series_data: List of info tuples for patches to add to the list,
-                         where each tuple contains the patch ID and a free-form
-                         patch date string.
+            series_data: List of info tuple of patches to add to the pending
+                         patches list.
 
         """
         sourceid = self.__get_sourceid(baseurl, project_id)
         tstamp = int(time.time())
 
         logging.debug("setting patches as pending: %s", series_data)
-        self.cur.executemany('INSERT OR REPLACE INTO '
-                             'pendingpatches(id, pdate, patchsource_id, '
-                             'timestamp) '
-                             'VALUES(?, ?, ?, ?)',
-                             [(patch_id, patch_date, sourceid, tstamp) for
-                              (patch_id, patch_date) in series_data])
+        self.cur.executemany(
+            'INSERT OR REPLACE INTO pendingpatches '
+            '(patch_id, timestamp) VALUES(?, ?)',
+            [(patch_id, tstamp) for (patch_id, patch_date) in series_data])
         self.conn.commit()
 
-    def __unset_patchset_pending(self, baseurl, patch_id_list):
-        """Remove a patch from the list of pending patches.
-
-        Remove each specified patch from the list of "pending" patches, for
-        the specified Patchwork base URL.
+    def __unset_patchset_pending(self, patch_id_list):
+        """Remove patches from the list of pending patches.
 
         Args:
-            baseurl:       Base URL of the Patchwork instance the patch IDs
-                           belong to.
             patch_id_list: List of IDs of patches to be removed from the list.
 
         """
         logging.debug("removing patches from pending list: %s", patch_id_list)
 
-        self.cur.executemany('DELETE FROM pendingpatches WHERE '
-                             'patchsource_id IN '
-                             '(SELECT DISTINCT id FROM patchsource WHERE '
-                             'baseurl = ?) '
-                             'AND id = ? ',
-                             [(baseurl, patch_id) for
-                              patch_id in patch_id_list])
+        self.cur.executemany(
+            'DELETE FROM pendingpatches WHERE patch_id = ?', [patches]
+        )
         self.conn.commit()
 
     def update_baseline(self, baserepo, commithash, commitdate,
@@ -469,13 +501,7 @@ def commit_tested(self, patches):
             patches:    List of patches that were tested
         """
         logging.debug("commit_tested: patches=%d", len(patches))
-        self.commit_series(patches)
-
-        for (patch_id, patch_name, patch_url, baseurl, project_id,
-             patch_date) in patches:
-            # TODO: Can accumulate per-project list instead of doing it one by
-            # one
-            self.__unset_patchset_pending(baseurl, [patch_id])
+        self.__unset_patchset_pending(patches)
 
     def __commit_testrun(self, result, buildid):
         """Add a test run to the database.

sktm/executable.py

@@ -47,6 +47,9 @@ def setup_parser():
     parser_baseline.add_argument("ref", type=str, help="Base repo ref to test")
     parser_baseline.set_defaults(func=cmd_baseline)
 
+    parser_checkpending = subparsers.add_parser("checkpending")
+    parser_checkpending.set_defaults(func=cmd_checkpending)
+
     parser_patchwork = subparsers.add_parser("patchwork")
     parser_patchwork.add_argument("repo", type=str, help="Base repo URL")
     parser_patchwork.add_argument("baseurl", type=str, help="Base URL")
@@ -92,6 +95,12 @@ def cmd_baseline(sw, cfg):
     sw.check_baseline()
 
 
+def cmd_checkpending(sw, cfg):
+    """Check any pending jobs that were started during the last run."""
+    logging.info("checking pending jobs")
+    sw.check_pending()
+
+
 def cmd_patchwork(sw, cfg):
     logging.info("checking patchwork: %s [%s]", cfg.get("baseurl"),
                  cfg.get("project"))
@@ -134,11 +143,6 @@ def main():
                       cfg.get("filter"), cfg.get("makeopts"))
 
     args.func(sw, cfg)
-    try:
-        sw.wait_for_pending()
-    except KeyboardInterrupt:
-        logging.info("Quitting...")
-        sw.cleanup()
 
 
 if __name__ == '__main__':

sktm/jenkins.py

@@ -332,6 +332,12 @@ def is_build_complete(self, jobname, buildid):
 
         return not build.is_running()
 
+    def get_build_status(self, jobname, buildid):
+        """Return the status of a Jenkins job."""
+        job = self.server.get_job(jobname)
+        build = job.get_build(buildid)
+        return build.get_status()
+
     def _params_eq(self, build, params):
         try:
             build_params = build.get_actions()["parameters"]