vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge master changes

Browse source
This commit is contained in:
Joseph Schorr 2015-02-05 13:11:16 -05:00
commit 9dfe523615
46 changed files with 1635 additions and 1232 deletions
Download patch file Download diff file Expand all files Collapse all files

13
data/database.py
View file

@ -29,6 +29,16 @@ SCHEME_RANDOM_FUNCTION = {
'postgresql+psycopg2': fn.Random,
}
def real_for_update(query):
return query.for_update()
def null_for_update(query):
return query
SCHEME_SPECIALIZED_FOR_UPDATE = {
'sqlite': null_for_update,
}
class CallableProxy(Proxy):
def __call__(self, *args, **kwargs):
if self.obj is None:
@ -68,6 +78,7 @@ class UseThenDisconnect(object):
db = Proxy()
read_slave = Proxy()
db_random_func = CallableProxy()
db_for_update = CallableProxy()
def validate_database_url(url, connect_timeout=5):
@ -105,6 +116,8 @@ def configure(config_object):
parsed_write_uri = make_url(write_db_uri)
db_random_func.initialize(SCHEME_RANDOM_FUNCTION[parsed_write_uri.drivername])
db_for_update.initialize(SCHEME_SPECIALIZED_FOR_UPDATE.get(parsed_write_uri.drivername,
real_for_update))
read_slave_uri = config_object.get('DB_READ_SLAVE_URI', None)
if read_slave_uri is not None:

20
data/model/legacy.py
View file

@ -14,7 +14,7 @@ from data.database import (User, Repository, Image, AccessToken, Role, Repositor
ExternalNotificationEvent, ExternalNotificationMethod,
RepositoryNotification, RepositoryAuthorizedEmail, TeamMemberInvite,
DerivedImageStorage, ImageStorageTransformation, random_string_generator,
db, BUILD_PHASE, QuayUserField, validate_database_url)
db, BUILD_PHASE, QuayUserField, validate_database_url, db_for_update)
from peewee import JOIN_LEFT_OUTER, fn
from util.validation import (validate_username, validate_email, validate_password,
INVALID_PASSWORD_MESSAGE)
@ -295,6 +295,9 @@ def delete_robot(robot_username):
def _list_entity_robots(entity_name):
""" Return the list of robots for the specified entity. This MUST return a query, not a
materialized list so that callers can use db_for_update.
"""
return (User
.select()
.join(FederatedLogin)
@ -903,14 +906,17 @@ def change_password(user, new_password):
delete_notifications_by_kind(user, 'password_required')
def change_username(user, new_username):
def change_username(user_id, new_username):
(username_valid, username_issue) = validate_username(new_username)
if not username_valid:
raise InvalidUsernameException('Invalid username %s: %s' % (new_username, username_issue))
with config.app_config['DB_TRANSACTION_FACTORY'](db):
# Reload the user for update
user = db_for_update(User.select().where(User.id == user_id)).get()
# Rename the robots
for robot in _list_entity_robots(user.username):
for robot in db_for_update(_list_entity_robots(user.username)):
_, robot_shortname = parse_robot_username(robot.username)
new_robot_name = format_robot_username(new_username, robot_shortname)
robot.username = new_robot_name
@ -1251,9 +1257,9 @@ def _find_or_link_image(existing_image, repository, username, translations, pref
storage.locations = {placement.location.name
for placement in storage.imagestorageplacement_set}
new_image = Image.create(docker_image_id=existing_image.docker_image_id,
repository=repository, storage=storage,
ancestors=new_image_ancestry)
new_image = Image.create(docker_image_id=existing_image.docker_image_id,
repository=repository, storage=storage,
ancestors=new_image_ancestry)
logger.debug('Storing translation %s -> %s', existing_image.id, new_image.id)
translations[existing_image.id] = new_image.id
@ -1403,7 +1409,7 @@ def set_image_metadata(docker_image_id, namespace_name, repository_name, created
Image.docker_image_id == docker_image_id))
try:
fetched = query.get()
fetched = db_for_update(query).get()
except Image.DoesNotExist:
raise DataModelException('No image with specified id and repository')

69
data/queue.py
View file

@ -1,6 +1,6 @@
from datetime import datetime, timedelta
from data.database import QueueItem, db
from data.database import QueueItem, db, db_for_update
from util.morecollections import AttrDict
@ -31,16 +31,24 @@ class WorkQueue(object):
QueueItem.processing_expires > now,
QueueItem.queue_name ** name_match_query))
def _available_jobs(self, now, name_match_query, running_query):
def _available_jobs(self, now, name_match_query):
return (QueueItem
.select()
.where(QueueItem.queue_name ** name_match_query, QueueItem.available_after <= now,
((QueueItem.available == True) | (QueueItem.processing_expires <= now)),
QueueItem.retries_remaining > 0, ~(QueueItem.queue_name << running_query)))
QueueItem.retries_remaining > 0))
def _available_jobs_not_running(self, now, name_match_query, running_query):
return (self
._available_jobs(now, name_match_query)
.where(~(QueueItem.queue_name << running_query)))
def _name_match_query(self):
return '%s%%' % self._canonical_name([self._queue_name] + self._canonical_name_match_list)
def _item_by_id_for_update(self, queue_id):
return db_for_update(QueueItem.select().where(QueueItem.id == queue_id)).get()
def update_metrics(self):
if self._reporter is None:
return
@ -52,7 +60,7 @@ class WorkQueue(object):
running_query = self._running_jobs(now, name_match_query)
running_count = running_query.distinct().count()
avialable_query = self._available_jobs(now, name_match_query, running_query)
avialable_query = self._available_jobs_not_running(now, name_match_query, running_query)
available_count = avialable_query.select(QueueItem.queue_name).distinct().count()
self._reporter(self._currently_processing, running_count, running_count + available_count)
@ -78,19 +86,26 @@ class WorkQueue(object):
def get(self, processing_time=300):
"""
Get an available item and mark it as unavailable for the default of five
minutes.
minutes. The result of this method must always be composed of simple
python objects which are JSON serializable for network portability reasons.
"""
now = datetime.utcnow()
name_match_query = self._name_match_query()
with self._transaction_factory(db):
running = self._running_jobs(now, name_match_query)
avail = self._available_jobs(now, name_match_query, running)
running = self._running_jobs(now, name_match_query)
avail = self._available_jobs_not_running(now, name_match_query, running)
item = None
try:
db_item = avail.order_by(QueueItem.id).get()
item = None
try:
db_item_candidate = avail.order_by(QueueItem.id).get()
with self._transaction_factory(db):
still_available_query = (db_for_update(self
._available_jobs(now, name_match_query)
.where(QueueItem.id == db_item_candidate.id)))
db_item = still_available_query.get()
db_item.available = False
db_item.processing_expires = now + timedelta(seconds=processing_time)
db_item.retries_remaining -= 1
@ -102,22 +117,22 @@ class WorkQueue(object):
})
self._currently_processing = True
except QueueItem.DoesNotExist:
self._currently_processing = False
except QueueItem.DoesNotExist:
self._currently_processing = False
# Return a view of the queue item rather than an active db object
return item
# Return a view of the queue item rather than an active db object
return item
def complete(self, completed_item):
with self._transaction_factory(db):
completed_item_obj = QueueItem.get(QueueItem.id == completed_item.id)
completed_item_obj = self._item_by_id_for_update(completed_item.id)
completed_item_obj.delete_instance()
self._currently_processing = False
def incomplete(self, incomplete_item, retry_after=300, restore_retry=False):
with self._transaction_factory(db):
retry_date = datetime.utcnow() + timedelta(seconds=retry_after)
incomplete_item_obj = QueueItem.get(QueueItem.id == incomplete_item.id)
incomplete_item_obj = self._item_by_id_for_update(incomplete_item.id)
incomplete_item_obj.available_after = retry_date
incomplete_item_obj.available = True
@ -127,16 +142,12 @@ class WorkQueue(object):
incomplete_item_obj.save()
self._currently_processing = False
@staticmethod
def extend_processing(queue_item_info, seconds_from_now, retry_count=None,
minimum_extension=MINIMUM_EXTENSION):
queue_item = QueueItem.get(QueueItem.id == queue_item_info.id)
new_expiration = datetime.utcnow() + timedelta(seconds=seconds_from_now)
def extend_processing(self, item, seconds_from_now, minimum_extension=MINIMUM_EXTENSION):
with self._transaction_factory(db):
queue_item = self._item_by_id_for_update(item.id)
new_expiration = datetime.utcnow() + timedelta(seconds=seconds_from_now)
# Only actually write the new expiration to the db if it moves the expiration some minimum
if new_expiration - queue_item.processing_expires > minimum_extension:
if retry_count is not None:
queue_item.retries_remaining = retry_count
queue_item.processing_expires = new_expiration
queue_item.save()
# Only actually write the new expiration to the db if it moves the expiration some minimum
if new_expiration - queue_item.processing_expires > minimum_extension:
queue_item.processing_expires = new_expiration
queue_item.save()