Fix queue handling to remove the dependency from repobuild, and have a cancel method

data/database.py

@@ -512,7 +512,7 @@ class RepositoryBuild(BaseModel):
   trigger = ForeignKeyField(RepositoryBuildTrigger, null=True, index=True)
   pull_robot = QuayUserField(null=True, related_name='buildpullrobot')
   logs_archived = BooleanField(default=False)
-  queue_item = ForeignKeyField(QueueItem, null=True, index=True)
+  queue_id = CharField(null=True, index=True)
 
 
 class LogEntryKind(BaseModel):

data/migrations/versions/707d5191eda_change_build_queue_reference_from_.py

@@ -0,0 +1,34 @@
+"""Change build queue reference from foreign key to an id.
+
+Revision ID: 707d5191eda
+Revises: 4ef04c61fcf9
+Create Date: 2015-02-23 12:36:33.814528
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '707d5191eda'
+down_revision = '4ef04c61fcf9'
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+def upgrade(tables):
+    ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('repositorybuild', sa.Column('queue_id', sa.String(length=255), nullable=True))
+    op.create_index('repositorybuild_queue_id', 'repositorybuild', ['queue_id'], unique=False)
+    op.drop_constraint(u'fk_repositorybuild_queue_item_id_queueitem', 'repositorybuild', type_='foreignkey')
+    op.drop_index('repositorybuild_queue_item_id', table_name='repositorybuild')
+    op.drop_column('repositorybuild', 'queue_item_id')
+    ### end Alembic commands ###
+
+
+def downgrade(tables):
+    ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('repositorybuild', sa.Column('queue_item_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=True))
+    op.create_foreign_key(u'fk_repositorybuild_queue_item_id_queueitem', 'repositorybuild', 'queueitem', ['queue_item_id'], ['id'])
+    op.create_index('repositorybuild_queue_item_id', 'repositorybuild', ['queue_item_id'], unique=False)
+    op.drop_index('repositorybuild_queue_id', table_name='repositorybuild')
+    op.drop_column('repositorybuild', 'queue_id')
+    ### end Alembic commands ###

data/model/legacy.py

@@ -2496,7 +2496,7 @@ def confirm_team_invite(code, user):
   found.delete_instance()
   return (team, inviter)
 
-def cancel_repository_build(build):
+def cancel_repository_build(build, work_queue):
   with config.app_config['DB_TRANSACTION_FACTORY'](db):
     # Reload the build for update.
     try:
@@ -2504,22 +2504,14 @@ def cancel_repository_build(build):
     except RepositoryBuild.DoesNotExist:
       return False
 
-    if build.phase != BUILD_PHASE.WAITING or not build.queue_item:
+    if build.phase != BUILD_PHASE.WAITING or not build.queue_id:
       return False
 
-    # Load the build queue item for update.
-    try:
-      queue_item = db_for_update(QueueItem.select()
-                                          .where(QueueItem.id == build.queue_item.id)).get()
-    except QueueItem.DoesNotExist:
+    # Try to cancel the queue item.
+    if not work_queue.cancel(build.queue_id):
       return False
 
-    # Check the queue item.
-    if not queue_item.available or queue_item.retries_remaining == 0:
-      return False
-
-    # Delete the queue item and build.
-    queue_item.delete_instance(recursive=True)
+    # Delete the build row.
     build.delete_instance()
     return True
 

data/queue.py

@@ -82,10 +82,19 @@ class WorkQueue(object):
     self._reporter(self._currently_processing, running_count,
                    running_count + available_not_running_count)
 
+  def has_retries_remaining(self, item_id):
+    """ Returns whether the queue item with the given id has any retries remaining. If the
+        queue item does not exist, returns False. """
+    with self._transaction_factory(db):
+      try:
+        return QueueItem.get(id=item_id).retries_remaining > 0
+      except QueueItem.DoesNotExist:
+        return False
+
   def put(self, canonical_name_list, message, available_after=0, retries_remaining=5):
     """
     Put an item, if it shouldn't be processed for some number of seconds,
-    specify that amount as available_after.
+    specify that amount as available_after. Returns the ID of the queue item added.
     """
 
     params = {
@@ -98,7 +107,7 @@ class WorkQueue(object):
     params['available_after'] = available_date
 
     with self._transaction_factory(db):
-      return QueueItem.create(**params)
+      return str(QueueItem.create(**params).id)
 
   def get(self, processing_time=300):
     """
@@ -141,10 +150,32 @@ class WorkQueue(object):
     # Return a view of the queue item rather than an active db object
     return item
 
+  def cancel(self, item_id):
+    """ Attempts to cancel the queue item with the given ID from the queue. Returns true on success
+        and false if the queue item could not be canceled. A queue item can only be canceled if
+        if is available and has retries remaining.
+    """
+
+    with self._transaction_factory(db):
+      # Load the build queue item for update.
+      try:
+        queue_item = db_for_update(QueueItem.select()
+                                            .where(QueueItem.id == item_id)).get()
+      except QueueItem.DoesNotExist:
+        return False
+
+      # Check the queue item.
+      if not queue_item.available or queue_item.retries_remaining == 0:
+        return False
+
+      # Delete the queue item.
+      queue_item.delete_instance(recursive=True)
+      return True
+
   def complete(self, completed_item):
     with self._transaction_factory(db):
       completed_item_obj = self._item_by_id_for_update(completed_item.id)
-      completed_item_obj.delete_instance()
+      completed_item_obj.delete_instance(recursive=True)
       self._currently_processing = False
 
   def incomplete(self, incomplete_item, retry_after=300, restore_retry=False):

endpoints/api/build.py

@@ -5,7 +5,7 @@ import datetime
 
 from flask import request, redirect
 
-from app import app, userfiles as user_files, build_logs, log_archive
+from app import app, userfiles as user_files, build_logs, log_archive, dockerfile_build_queue
 from endpoints.api import (RepositoryParamResource, parse_args, query_param, nickname, resource,
                            require_repo_read, require_repo_write, validate_json_request,
                            ApiResource, internal_only, format_date, api, Unauthorized, NotFound,
@@ -79,7 +79,8 @@ def build_status_view(build_obj, can_write=False):
   # If the phase is internal error, return 'error' instead of the number if retries
   # on the queue item is 0.
   if phase == database.BUILD_PHASE.INTERNAL_ERROR:
-    if build_obj.queue_item is None or build_obj.queue_item.retries_remaining == 0:
+    retry = build_obj.queue_id and dockerfile_build_queue.has_retries_remaining(build_obj.queue_id)
+    if not retry:
       phase = database.BUILD_PHASE.ERROR
 
   logger.debug('Can write: %s job_config: %s', can_write, build_obj.job_config)
@@ -226,7 +227,7 @@ class RepositoryBuildResource(RepositoryParamResource):
     if build.repository.name != repository or build.repository.namespace_user.username != namespace:
       raise NotFound()
 
-    if model.cancel_repository_build(build):
+    if model.cancel_repository_build(build, dockerfile_build_queue):
       return 'Okay', 201
     else:
       raise InvalidRequest('Build is currently running or has finished')

endpoints/common.py

@@ -237,11 +237,11 @@ def start_build(repository, dockerfile_id, tags, build_name, subdir, manual,
       'pull_credentials': model.get_pull_credentials(pull_robot_name) if pull_robot_name else None
     })
 
-    queue_item = dockerfile_build_queue.put([repository.namespace_user.username, repository.name],
-                                            json_data,
-                                            retries_remaining=3)
+    queue_id = dockerfile_build_queue.put([repository.namespace_user.username, repository.name],
+                                          json_data,
+                                          retries_remaining=3)
 
-    build_request.queue_item = queue_item
+    build_request.queue_id = queue_id
     build_request.save()
 
   # Add the build to the repo's log.

test/test_api_usage.py

@@ -1331,6 +1331,13 @@ class TestRepositoryBuildResource(ApiTestCase):
     self.assertEquals(1, len(json['builds']))
     self.assertEquals(uuid, json['builds'][0]['id'])
 
+    # Find the build's queue item.
+    build_ref = database.RepositoryBuild.get(uuid=uuid)
+    queue_item = database.QueueItem.get(id=build_ref.queue_id)
+
+    self.assertTrue(queue_item.available)
+    self.assertTrue(queue_item.retries_remaining > 0)
+
     # Cancel the build.
     self.deleteResponse(RepositoryBuildResource,
                         params=dict(repository=ADMIN_ACCESS_USER + '/simple', build_uuid=uuid),
@@ -1342,6 +1349,12 @@ class TestRepositoryBuildResource(ApiTestCase):
 
     self.assertEquals(0, len(json['builds']))
 
+    # Check for the build's queue item.
+    try:
+      database.QueueItem.get(id=build_ref.queue_id)
+      self.fail('QueueItem still exists for build')
+    except database.QueueItem.DoesNotExist:
+      pass
 
   def test_attemptcancel_scheduledbuild(self):
     self.login(ADMIN_ACCESS_USER)