Merge remote-tracking branch 'upstream/master'

lago.spec.in

@@ -50,7 +50,6 @@ BuildArch: noarch
 BuildRequires: python2-devel
 BuildRequires: python-stevedore
 BuildRequires: python-setuptools
-BuildRequires: python-lockfile
 BuildRequires: python-yaml
 BuildRequires: pyxdg
 BuildRequires: python-xmltodict
@@ -90,7 +89,6 @@ Requires: libvirt >= 1.2.8
 Requires: libvirt-python
 Requires: python-libguestfs >= 1.30
 Requires: python-lxml
-Requires: python-lockfile
 Requires: python-pbr
 Requires: python-xmltodict
 Requires: python-scp

lago/plugins/__init__.py

@@ -22,6 +22,7 @@
 """
 from stevedore import ExtensionManager
 import logging
+import warnings
 
 LOGGER = logging.getLogger(__name__)
 
@@ -34,6 +35,12 @@
     'vm-provider': 'lago.plugins.vm_provider',
 }
 
+# Warnings that are emitted by stevedore package and we wnat to ignore.
+STEVEDORE_WARN_MSG = {
+    'Parameters to load are deprecated.  '
+    'Call .resolve and .require separately.',
+}
+
 
 class PluginError(Exception):
     pass
@@ -51,6 +58,17 @@ class Plugin(object):
 
 
 def load_plugins(namespace, instantiate=True):
+    with warnings.catch_warnings(record=True) as wlist:
+        plugins = _load_plugins(namespace, instantiate)
+        for warn in wlist:
+            msg = str(warn.message)
+            if msg not in STEVEDORE_WARN_MSG:
+                LOGGER.warning(msg)
+
+    return plugins
+
+
+def _load_plugins(namespace, instantiate=True):
     """
     Loads all the plugins for the given namespace
 

lago/prefix.py

@@ -168,6 +168,8 @@ def _create_ssh_keys(self):
                 'ssh-keygen',
                 '-t',
                 'rsa',
+                '-m',
+                'PEM',
                 '-N',
                 '',
                 '-f',

lago/subnet_lease.py

@@ -26,14 +26,13 @@
 import logging
 from netaddr import IPNetwork, AddrFormatError
 from textwrap import dedent
-from lockfile.mkdirlockfile import LockFailed
 
 from .config import config
 from . import utils, log_utils
 
 LOGGER = logging.getLogger(__name__)
 LogTask = functools.partial(log_utils.LogTask, logger=LOGGER)
-LOCK_NAME = 'leases'
+LOCK_NAME = 'subnet-lease.lock'
 
 
 class SubnetStore(object):
@@ -148,7 +147,7 @@ def acquire(self, uuid_path, subnet=None):
                     acquired_subnet = self._acquire(uuid_path)
 
                 return acquired_subnet
-        except (utils.TimerException, LockFailed):
+        except (utils.TimerException, IOError):
             raise LagoSubnetLeaseLockException(self.path)
 
     def _acquire(self, uuid_path):
@@ -269,7 +268,7 @@ def list_leases(self, uuid=None):
 
         leases = [
             self.create_lease_object_from_idx(lease_file.split('.')[0])
-            for lease_file in lease_files
+            for lease_file in lease_files if lease_file != LOCK_NAME
         ]
         if not uuid:
             return leases
@@ -301,7 +300,7 @@ def release(self, subnets):
             with self._create_lock():
                 for subnet in subnets_iter:
                     self._release(self.create_lease_object_from_subnet(subnet))
-        except (utils.TimerException, LockFailed):
+        except (utils.TimerException, IOError):
             raise LagoSubnetLeaseLockException(self.path)
 
     def _release(self, lease):

lago/templates.py

@@ -19,7 +19,6 @@
 
 """
 import errno
-import functools
 import json
 import logging
 import os
@@ -28,8 +27,6 @@
 import urllib
 import sys
 
-import lockfile
-
 import utils
 from . import log_utils
 from .config import config
@@ -513,19 +510,6 @@ def download(self, destination):
         self._source.download_image(self._handle, destination)
 
 
-def _locked(func):
-    """
-    Decorator that ensures that the decorated function has the lock of the
-    repo while running, meant to decorate only bound functions for classes that
-    have `lock_path` method.
-    """
-
-    @functools.wraps(func)
-    def wrapper(self, *args, **kwargs):
-        with lockfile.LockFile(self.lock_path()):
-            return func(self, *args, **kwargs)
-
-
 class TemplateStore:
     """
     Local cache to store templates
@@ -635,7 +619,7 @@ def download(self, temp_ver, store_metadata=True):
         dest = self._prefixed(temp_ver.name)
         temp_dest = '%s.tmp' % dest
 
-        with lockfile.LockFile(dest):
+        with utils.LockFile(dest + '.lock'):
             # Image was downloaded while we were waiting
             if os.path.exists(dest):
                 return

lago/utils.py

@@ -35,7 +35,6 @@
 import yaml
 import pkg_resources
 from io import StringIO
-import lockfile
 import argparse
 import configparser
 import uuid as uuid_m
@@ -367,21 +366,63 @@ def stop(self):
         signal.alarm(0)
 
 
+class Flock(object):
+    """A wrapper class around flock
+
+    Attributes:
+        path(str): Path to the lock file
+        readonly(bool): If true create a shared lock, otherwise
+            create an exclusive lock.
+        blocking(bool) If true block the calling process if the
+            lock is already acquired.
+    """
+
+    def __init__(self, path, readonly=False, blocking=True):
+        self._path = path
+        self._fd = None
+        if readonly:
+            self._op = fcntl.LOCK_SH
+        else:
+            self._op = fcntl.LOCK_EX
+
+        if not blocking:
+            self._op |= fcntl.LOCK_NB
+
+    def acquire(self):
+        """Acquire the lock
+
+        Raises:
+            IOError: if the call to flock fails
+        """
+        self._fd = open(self._path, mode='w+')
+        fcntl.flock(self._fd, self._op)
+
+    def release(self):
+        self._fd.close()
+
+
 class LockFile(object):
     """
-    Context manager that creates a lock around a directory, with optional
-    timeout in the acquire operation
+    Context manager that creates a file based lock, with optional
+    timeout in the acquire operation.
+
+    This context manager should be used only from the main Thread.
 
     Args:
         path(str): path to the dir to lock
         timeout(int): timeout in seconds to wait while acquiring the lock
-        **kwargs(dict): Any other param to pass to `lockfile.LockFile`
+        lock_cls(callable): A callable which returns a Lock object that
+            implements the acquire and release methods.
+            The default is Flock.
+        **kwargs(dict): Any other param to pass to the `lock_cls` instance.
+
     """
 
-    def __init__(self, path, timeout=None, **kwargs):
+    def __init__(self, path, timeout=None, lock_cls=None, **kwargs):
         self.path = path
         self.timeout = timeout or 0
-        self.lock = lockfile.LockFile(path=path, **kwargs)
+        self._lock_cls = lock_cls or Flock
+        self.lock = self._lock_cls(path=path, **kwargs)
 
     def __enter__(self):
         """
@@ -403,10 +444,9 @@ def __enter__(self):
             )
 
     def __exit__(self, *_):
-        if self.lock.is_locked():
-            LOGGER.debug('Trying to release lock for {}'.format(self.path))
-            self.lock.release()
-            LOGGER.debug('Lock for {} was released'.format(self.path))
+        LOGGER.debug('Trying to release lock for {}'.format(self.path))
+        self.lock.release()
+        LOGGER.debug('Lock for {} was released'.format(self.path))
 
 
 def read_nonblocking(file_descriptor):

requirements.txt

@@ -1,7 +1,6 @@
 configparser
 enum34
 libvirt-python
-lockfile
 lxml
 paramiko
 pbr

test-requires.txt

@@ -1,6 +1,6 @@
 -r requirements.txt
 dulwich
-flake8
+flake8==3.5.0
 mock
 futures
 xmlunittest

tests/functional-sdk/conftest.py

@@ -8,15 +8,15 @@
 
 _local_config = {
     'check_patch': {
-        'images': ['el7.4-base']
+        'images': ['el7.5-base']
     },
     'check_merged':
         {
             'images': [
-                'el7.4-base-1',
+                'el7.5-base',
                 'el6-base',
-                'fc26-base',
-                'fc27-base',
+                'fc28-base',
+                'fc29-base',
             ]
         }  # noqa: E123
 }

tests/functional-sdk/test_sdk_sanity.py

@@ -5,6 +5,8 @@
 import logging
 import uuid
 import filecmp
+from time import sleep
+
 from jinja2 import Environment, BaseLoader
 from lago import sdk
 from lago.utils import run_command
@@ -217,14 +219,22 @@ def test_ansible_inventory(monkeypatch, env, test_results, vms):
 
 def test_systemd_analyze(test_results, vms, vm_name):
     vm = vms[vm_name]
+    retries = 3
+
     res = vm.ssh(['command', '-v', 'systemd-analyze'])
     if res.code != 0:
         raise pytest.skip(
             'systemd-analyze not available on {0}'.format(vm_name)
         )
 
-    res = vm.ssh(['systemd-analyze'])
-    assert res.code == 0
+    for i in xrange(retries):
+        res = vm.ssh(['systemd-analyze'])
+        if not res:
+            break
+        sleep(3)
+    else:
+        pytest.fail('Failed to run systemd-analyze on {}'.format(vm_name))
+
     log = '\n'.join([res.out, res.err])
 
     res = vm.ssh(['systemd-analyze', 'blame'])

tests/unit/lago/test_lock.py

@@ -0,0 +1,74 @@
+import signal
+from multiprocessing import Process, Event
+from os import WNOHANG, kill, waitpid
+from time import sleep
+
+import pytest
+
+from lago.utils import LockFile, TimerException
+
+
+def lock_path(run_path, duration, event=None):
+    with LockFile(run_path):
+        if event:
+            event.set()
+        sleep(duration)
+
+
+class ProcessWrapper(object):
+    def __init__(self, daemon=True, **kwargs):
+        self._p = Process(**kwargs)
+        self._p.daemon = daemon
+
+    def __getattr__(self, name):
+        return getattr(self._p, name)
+
+    def kill(self, sig=None):
+        sig = signal.SIGKILL if sig is None else sig
+        kill(self.pid, sig)
+
+    def waitpid(self):
+        return waitpid(self.pid, WNOHANG)
+
+    def __enter__(self):
+        self.start()
+
+    def __exit__(self, *args):
+        self.kill()
+
+
+@pytest.fixture
+def lockfile(tmpdir):
+    return str(tmpdir.join('test-lock'))
+
+
+@pytest.fixture
+def event():
+    return Event()
+
+
+@pytest.fixture
+def p_wrapper(lockfile, event):
+    duration = 60
+    event.clear()
+
+    return ProcessWrapper(target=lock_path, args=(lockfile, duration, event))
+
+
+def test_should_fail_to_lock_when_already_locked(lockfile, p_wrapper, event):
+    with p_wrapper:
+        assert event.wait(30), 'Timeout while waiting for child process'
+        with pytest.raises(TimerException), LockFile(lockfile, timeout=1):
+            pass
+
+
+def test_should_succeed_to_lock_a_stale_lock(lockfile, p_wrapper, event):
+    p_wrapper.start()
+    assert event.wait(30), 'Timeout while waiting for child process'
+
+    p_wrapper.kill()
+    # If the process is still running waitpid returns (0, 0)
+    assert not any(p_wrapper.waitpid()), 'Failed to kill child process'
+
+    with LockFile(lockfile, timeout=1):
+        pass