Merge pull request #289 from aplanas/DimStar77-locking

Dim star77 locking

osc-check_repo.py

@@ -26,10 +26,9 @@ import tempfile
 import traceback
 import sys
 
-import osc
-from osc import oscerr
 from osc import cmdln
-
+from osc import conf
+from osc import oscerr
 
 # Expand sys.path to search modules inside the pluging directory
 PLUGINDIR = os.path.dirname(os.path.realpath(__file__.replace('.pyc', '.py')))
@@ -533,7 +532,7 @@ def do_check_repo(self, subcmd, opts, *args):
             self._check_repo_group(id_, reqs, debug=opts.verbose)
         except Exception as e:
             print 'ERROR -- An exception happends while checking a group [%s]' % e
-            if osc.conf.config['debug']:
+            if conf.config['debug']:
                 print traceback.format_exc()
         print
         print

osc-staging.py

@@ -11,17 +11,18 @@ import os
 import os.path
 import sys
 
-from osc import cmdln, oscerr
+from osc import cmdln
+from osc import oscerr
 
 # Expand sys.path to search modules inside the pluging directory
 _plugin_dir = os.path.expanduser('~/.osc-plugins')
 sys.path.append(_plugin_dir)
-
 from osclib.accept_command import AcceptCommand
 from osclib.check_command import CheckCommand
 from osclib.cleanup_rings import CleanupRings
 from osclib.freeze_command import FreezeCommand
 from osclib.list_command import ListCommand
+from osclib.obslock import OBSLock
 from osclib.select_command import SelectCommand
 from osclib.stagingapi import StagingAPI
 from osclib.unselect_command import UnselectCommand
@@ -105,32 +106,34 @@ def do_staging(self, subcmd, opts, *args):
     # init the obs access
     opts.apiurl = self.get_api_url()
     opts.verbose = False
-    api = StagingAPI(opts.apiurl, opts.project)
 
-    # call the respective command and parse args by need
+    with OBSLock(opts.apiurl, 'openSUSE:%s:Staging' % opts.project):
-    if cmd == 'check':
+        api = StagingAPI(opts.apiurl, opts.project)
-        prj = args[1] if len(args) > 1 else None
+
-        CheckCommand(api).perform(prj, opts.old)
+        # call the respective command and parse args by need
-    elif cmd == 'freeze':
+        if cmd == 'check':
-        for prj in args[1:]:
+            prj = args[1] if len(args) > 1 else None
-            FreezeCommand(api).perform(api.prj_from_letter(prj))
+            CheckCommand(api).perform(prj, opts.old)
-    elif cmd == 'accept':
+        elif cmd == 'freeze':
-        cmd = AcceptCommand(api)
+            for prj in args[1:]:
-        for prj in args[1:]:
+                FreezeCommand(api).perform(api.prj_from_letter(prj))
-            if not cmd.perform(api.prj_from_letter(prj)):
+        elif cmd == 'accept':
-                return
+            cmd = AcceptCommand(api)
-        cmd.accept_other_new()
+            for prj in args[1:]:
-        cmd.update_factory_version()
+                if not cmd.perform(api.prj_from_letter(prj)):
-        cmd.sync_buildfailures()
+                    return
-    elif cmd == 'unselect':
+            cmd.accept_other_new()
-        UnselectCommand(api).perform(args[1:])
+            cmd.update_factory_version()
-    elif cmd == 'select':
+            cmd.sync_buildfailures()
-        tprj = api.prj_from_letter(args[1])
+        elif cmd == 'unselect':
-        if opts.add:
+            UnselectCommand(api).perform(args[1:])
-            api.mark_additional_packages(tprj, [opts.add])
+        elif cmd == 'select':
-        else:
+            tprj = api.prj_from_letter(args[1])
-            SelectCommand(api).perform(tprj, args[2:], opts.move, opts.from_)
+            if opts.add:
-    elif cmd == 'cleanup_rings':
+                api.mark_additional_packages(tprj, [opts.add])
-        CleanupRings(api).perform()
+            else:
-    elif cmd == 'list':
+                SelectCommand(api).perform(tprj, args[2:], opts.move, opts.from_)
-        ListCommand(api).perform()
+        elif cmd == 'cleanup_rings':
+            CleanupRings(api).perform()
+        elif cmd == 'list':
+            ListCommand(api).perform()

osclib/obslock.py

@@ -0,0 +1,99 @@
+# Copyright (C) 2015 SUSE Linux Products GmbH
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+from datetime import datetime
+import time
+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):
+        self.apiurl = apiurl
+        self.project = project
+        # TTL is measured in seconds
+        self.ttl = ttl
+        self.user = conf.config['api_host_options'][apiurl]['user']
+        self.locked = False
+
+    def _signature(self):
+        """Create a signature with a timestamp."""
+        return '%s@%s' % (self.user, datetime.isoformat(datetime.utcnow()))
+
+    def _parse(self, signature):
+        """Parse a signature into an user and a timestamp."""
+        user, ts = None, None
+        try:
+            user, ts_str = signature.split('@')
+            ts = datetime.strptime(ts_str, '%Y-%m-%dT%H:%M:%S.%f')
+        except (AttributeError, ValueError):
+            pass
+        return user, ts
+
+    def _read(self):
+        url = makeurl(self.apiurl, ['source', self.project, '_attribute', 'openSUSE:LockedBy'])
+        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.project, '_attribute', 'openSUSE:LockedBy'])
+        data = """
+        <attributes>
+          <attribute namespace='openSUSE' name='LockedBy'>
+            <value>%s</value>
+          </attribute>
+        </attributes>""" % signature
+        http_POST(url, data=data)
+
+    def acquire(self):
+        user, 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))
+            if user != self.user and (now - ts).seconds < self.ttl:
+                print 'Lock acquired by [%s]. Try later.' % user
+                exit(-1)
+                # raise Exception('Lock acquired by [%s]. Try later.' % user)
+        self._write(self._signature())
+
+        time.sleep(1)
+        user, ts = self._parse(self._read())
+        if user != self.user:
+            raise Exception('Race condition, [%s] wins. Try later.' % user)
+
+        return self
+
+    def release(self):
+        user, ts = self._parse(self._read())
+        if user == self.user:
+            self._write('')
+
+    __enter__ = acquire
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.release()

osclib/stagingapi.py

@@ -46,6 +46,7 @@ class StagingAPI(object):
         self.ring_packages = self._generate_ring_packages()
         self.packages_staged = self._get_staged_requests()
 
+
     def makeurl(self, l, query=None):
         """
         Wrapper around osc's makeurl passing our apiurl
@@ -156,7 +157,7 @@ class StagingAPI(object):
         """
 
         filelist = []
-        query ={
+        query = {
             'extension': extension
         }
 
@@ -953,7 +954,7 @@ class StagingAPI(object):
         specfile = self.load_file_content(project, package, '{}.spec'.format(package))
         if specfile:
             try:
-                version = re.findall("^Version:(.*)",specfile,re.MULTILINE)[0].strip()
+                version = re.findall('^Version:(.*)', specfile, re.MULTILINE)[0].strip()
             except IndexError:
                 pass
         return version
@@ -1061,7 +1062,10 @@ class StagingAPI(object):
         return pkglist
 
     def rebuild_pkg(self, package, prj, arch, code=None):
-        query = { 'cmd': 'rebuild', 'arch': arch }
+        query = {
+            'cmd': 'rebuild',
+            'arch': arch
+        }
         if package:
             query['package'] = package
         pkg = query['package']
@@ -1070,7 +1074,6 @@ class StagingAPI(object):
 
         try:
             print "tried to trigger rebuild for project '%s' package '%s'" % (prj, pkg)
-            f = http_POST(u)
+            http_POST(u)
-
         except:
             print "could not trigger rebuild for project '%s' package '%s'" % (prj, pkg)