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

Merge branch 'master' into v2

Browse source
This commit is contained in:
Joseph Schorr 2015-02-05 15:37:14 -05:00
commit cf774e23df
74 changed files with 2123 additions and 1433 deletions
Download patch file Download diff file Expand all files Collapse all files

36
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,15 @@ class UseThenDisconnect(object):
db = Proxy()
read_slave = Proxy()
db_random_func = CallableProxy()
db_for_update = CallableProxy()
def validate_database_url(url, connect_timeout=5):
driver = _db_from_url(url, {
'connect_timeout': connect_timeout
})
driver.connect()
driver.close()
def _db_from_url(url, db_kwargs):
@ -82,6 +101,10 @@ def _db_from_url(url, db_kwargs):
if parsed_url.password:
db_kwargs['password'] = parsed_url.password
# Note: sqlite does not support connect_timeout.
if parsed_url.drivername == 'sqlite' and 'connect_timeout' in db_kwargs:
del db_kwargs['connect_timeout']
return SCHEME_DRIVERS[parsed_url.drivername](parsed_url.database, **db_kwargs)
@ -93,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:
@ -122,8 +147,9 @@ def close_db_filter(_):
class QuayUserField(ForeignKeyField):
def __init__(self, allows_robots=False, *args, **kwargs):
def __init__(self, allows_robots=False, robot_null_delete=False, *args, **kwargs):
self.allows_robots = allows_robots
self.robot_null_delete = robot_null_delete
if not 'rel_model' in kwargs:
kwargs['rel_model'] = User
@ -157,7 +183,11 @@ class User(BaseModel):
for query, fk in self.dependencies(search_nullable=True):
if isinstance(fk, QuayUserField) and fk.allows_robots:
model = fk.model_class
model.delete().where(query).execute()
if fk.robot_null_delete:
model.update(**{fk.name: None}).where(query).execute()
else:
model.delete().where(query).execute()
# Delete the instance itself.
super(User, self).delete_instance(recursive=False, delete_nullable=False)
@ -477,7 +507,7 @@ class LogEntry(BaseModel):
kind = ForeignKeyField(LogEntryKind, index=True)
account = QuayUserField(index=True, related_name='account')
performer = QuayUserField(allows_robots=True, index=True, null=True,
related_name='performer')
related_name='performer', robot_null_delete=True)
repository = ForeignKeyField(Repository, index=True, null=True)
datetime = DateTimeField(default=datetime.now, index=True)
ip = CharField(null=True)

35
data/model/legacy.py
View file

@ -15,7 +15,7 @@ from data.database import (User, Repository, Image, AccessToken, Role, Repositor
RepositoryNotification, RepositoryAuthorizedEmail, TeamMemberInvite,
DerivedImageStorage, ImageStorageTransformation, random_string_generator,
db, BUILD_PHASE, QuayUserField, ImageStorageSignature,
ImageStorageSignatureKind)
ImageStorageSignatureKind, 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)
@ -296,6 +296,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)
@ -904,14 +907,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
@ -1252,9 +1258,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
@ -1433,7 +1439,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')
@ -2287,11 +2293,20 @@ def delete_user(user):
# TODO: also delete any repository data associated
def check_health():
def check_health(app_config):
# Attempt to connect to the database first. If the DB is not responding,
# using the validate_database_url will timeout quickly, as opposed to
# making a normal connect which will just hang (thus breaking the health
# check).
try:
validate_database_url(app_config['DB_URI'], connect_timeout=3)
except Exception:
logger.exception('Could not connect to the database')
return False
# We will connect to the db, check that it contains some log entry kinds
try:
found_count = LogEntryKind.select().count()
return found_count > 0
return bool(list(LogEntryKind.select().limit(1)))
except:
return False

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()