Split pending job checking from queueing

Alter the pendingpatches table so that it only contains a timestamp
and a patch_id that links to the id field of the patch table. All
of the relevant data about a patch is in the patch table already and
there is no need to replicate it in pendingpatches.

Change check_pending() so that it searches the pendingjobs table and
takes action on completed baselines and patchwork tests.

Handle any aborted jobs properly without throwing exceptions.

Signed-off-by: Major Hayden <[email protected]>

Author: major

db_migrations/04-alter-pendingpatches.sql

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

sktm/__init__.py

@@ -336,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,9 @@ 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)
+                  FOREIGN KEY(patch_id) REFERENCES patch(id)
                 );
 
                 CREATE TABLE pendingjobs(
@@ -425,57 +424,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,
@@ -521,13 +499,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

@@ -143,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__':