6069245350
Distinct copyrights were left as I do not wish to track down commit history to ensure it properly documents the copyright holders. Also left non-GPLv2 licenses and left bs_copy untouched as a mirror from OBS. Already have a mix of with and without headers and even OBS does not place on majority of files. If SUSE lawyers have an issue it will come up in legal review for Factory.
146 lines
4.9 KiB
Python
146 lines
4.9 KiB
Python
from datetime import datetime
|
|
import time
|
|
import warnings
|
|
from xml.etree import cElementTree as ET
|
|
|
|
from osc import conf
|
|
from osc.core import makeurl
|
|
from osc.core import http_GET
|
|
from osc.core import http_POST
|
|
|
|
|
|
class OBSLock(object):
|
|
"""Implement a distributed lock using a shared OBS resource."""
|
|
|
|
def __init__(self, apiurl, project, ttl=3600, reason=None, needed=True):
|
|
self.apiurl = apiurl
|
|
self.project = project
|
|
self.lock = conf.config[project]['lock']
|
|
self.ns = conf.config[project]['lock-ns']
|
|
# TTL is measured in seconds
|
|
self.ttl = ttl
|
|
self.user = conf.config['api_host_options'][apiurl]['user']
|
|
self.reason = reason
|
|
self.reason_sub = None
|
|
self.locked = False
|
|
self.needed = needed
|
|
|
|
def _signature(self):
|
|
"""Create a signature with a timestamp."""
|
|
reason = str(self.reason)
|
|
if self.reason_sub:
|
|
reason += ' ({})'.format(self.reason_sub)
|
|
reason = reason.replace('@', 'at').replace('#', 'hash')
|
|
return '%s#%s@%s' % (self.user, reason, datetime.isoformat(datetime.utcnow()))
|
|
|
|
def _parse(self, signature):
|
|
"""Parse a signature into an user and a timestamp."""
|
|
user, reason, reason_sub, ts = None, None, None, None
|
|
try:
|
|
rest, ts_str = signature.split('@')
|
|
user, reason = rest.split('#')
|
|
if ' (hold' in reason:
|
|
reason, reason_sub = reason.split(' (', 1)
|
|
reason_sub = reason_sub.rstrip(')')
|
|
ts = datetime.strptime(ts_str, '%Y-%m-%dT%H:%M:%S.%f')
|
|
except (AttributeError, ValueError):
|
|
pass
|
|
return user, reason, reason_sub, ts
|
|
|
|
def _read(self):
|
|
url = makeurl(self.apiurl, ['source', self.lock, '_attribute', '%s:LockedBy' % self.ns])
|
|
root = ET.parse(http_GET(url)).getroot()
|
|
signature = None
|
|
try:
|
|
signature = root.find('.//value').text
|
|
except (AttributeError, ValueError):
|
|
pass
|
|
return signature
|
|
|
|
def _write(self, signature):
|
|
url = makeurl(self.apiurl, ['source', self.lock, '_attribute'])
|
|
data = """
|
|
<attributes>
|
|
<attribute namespace='%s' name='LockedBy'>
|
|
<value>%s</value>
|
|
</attribute>
|
|
</attributes>""" % (self.ns, signature)
|
|
http_POST(url, data=data)
|
|
|
|
def acquire(self):
|
|
if not self.needed: return self
|
|
|
|
# If the project doesn't have locks configured, raise a
|
|
# Warning (but continue the operation)
|
|
if not self.lock:
|
|
warnings.warn('Locking attribute is not found. Create one to avoid race conditions.')
|
|
return self
|
|
|
|
user, reason, reason_sub, ts = self._parse(self._read())
|
|
if user and ts:
|
|
now = datetime.utcnow()
|
|
if now < ts:
|
|
raise Exception('Lock acquired from the future [%s] by [%s]. Try later.' % (ts, user))
|
|
delta = now - ts
|
|
if delta.total_seconds() < self.ttl:
|
|
# Existing lock that has not expired.
|
|
stop = True
|
|
if user == self.user:
|
|
if reason.startswith('hold'):
|
|
# Command being issued during a hold.
|
|
self.reason_sub = reason
|
|
stop = False
|
|
elif reason == 'lock':
|
|
# Second pass to acquire hold.
|
|
stop = False
|
|
|
|
if stop:
|
|
print 'Lock acquired by [%s] %s ago, reason <%s>. Try later.' % (user, delta, reason)
|
|
exit(-1)
|
|
self._write(self._signature())
|
|
|
|
time.sleep(1)
|
|
user, _, _, _ = self._parse(self._read())
|
|
if user != self.user:
|
|
raise Exception('Race condition, [%s] wins. Try later.' % user)
|
|
self.locked = True
|
|
|
|
return self
|
|
|
|
def release(self, force=False):
|
|
if not force and not self.needed: return
|
|
|
|
# If the project do not have locks configured, simply ignore
|
|
# the operation.
|
|
if not self.lock:
|
|
return
|
|
|
|
user, reason, reason_sub, _ = self._parse(self._read())
|
|
clear = False
|
|
if user == self.user:
|
|
if reason_sub:
|
|
self.reason = reason_sub
|
|
self.reason_sub = None
|
|
self._write(self._signature())
|
|
elif not reason.startswith('hold') or force:
|
|
# Only clear a command lock as hold can only be cleared by force.
|
|
clear = True
|
|
elif user is not None and force:
|
|
# Clear if a lock is present and force.
|
|
clear = True
|
|
|
|
if clear:
|
|
self._write('')
|
|
self.locked = False
|
|
|
|
def hold(self, message=None):
|
|
self.reason = 'hold'
|
|
if message:
|
|
self.reason += ': ' + message
|
|
self.acquire()
|
|
|
|
__enter__ = acquire
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
self.release()
|