Wait for apt lock (#1034)

Currently any attempt to run an apt command while another process holds
an apt lock will fail. We should instead wait to acquire the apt lock.

LP: #1944611

cloudinit/distros/debian.py

@@ -7,8 +7,9 @@
 # Author: Joshua Harlow <harlowja@yahoo-inc.com>
 #
 # This file is part of cloud-init. See LICENSE file for license information.
-
+import fcntl
 import os
+import time
 
 from cloudinit import distros
 from cloudinit import helpers
@@ -22,6 +23,7 @@
 
 LOG = logging.getLogger(__name__)
 
+APT_LOCK_WAIT_TIMEOUT = 30
 APT_GET_COMMAND = ('apt-get', '--option=Dpkg::Options::=--force-confold',
                    '--option=Dpkg::options::=--force-unsafe-io',
                    '--assume-yes', '--quiet')
@@ -41,6 +43,12 @@
 NETWORK_CONF_FN = "/etc/network/interfaces.d/50-cloud-init"
 LOCALE_CONF_FN = "/etc/default/locale"
 
+APT_LOCK_FILES = [
+    '/var/lib/dpkg/lock',
+    '/var/lib/apt/lists/lock',
+    '/var/cache/apt/archives/lock',
+]
+
 
 class Distro(distros.Distro):
     hostname_conf_fn = "/etc/hostname"
@@ -155,7 +163,78 @@ def _get_localhost_ip(self):
     def set_timezone(self, tz):
         distros.set_etc_timezone(tz=tz, tz_file=self._find_tz_file(tz))
 
+    def _apt_lock_available(self, lock_files=None):
+        """Determines if another process holds any apt locks.
+
+        If all locks are clear, return True else False.
+        """
+        if lock_files is None:
+            lock_files = APT_LOCK_FILES
+        for lock in lock_files:
+            if not os.path.exists(lock):
+                # Only wait for lock files that already exist
+                continue
+            with open(lock, 'w') as handle:
+                try:
+                    fcntl.lockf(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                except OSError:
+                    return False
+        return True
+
+    def _wait_for_apt_command(
+        self, short_cmd, subp_kwargs, timeout=APT_LOCK_WAIT_TIMEOUT
+    ):
+        """Wait for apt install to complete.
+
+        short_cmd: Name of command like "upgrade" or "install"
+        subp_kwargs: kwargs to pass to subp
+        """
+        start_time = time.time()
+        LOG.debug('Waiting for apt lock')
+        while time.time() - start_time < timeout:
+            if not self._apt_lock_available():
+                time.sleep(1)
+                continue
+            LOG.debug('apt lock available')
+            try:
+                # Allow the output of this to flow outwards (not be captured)
+                log_msg = "apt-%s [%s]" % (
+                    short_cmd,
+                    ' '.join(subp_kwargs['args'])
+                )
+                return util.log_time(
+                    logfunc=LOG.debug,
+                    msg=log_msg,
+                    func=subp.subp,
+                    kwargs=subp_kwargs,
+                )
+            except subp.ProcessExecutionError:
+                # Even though we have already waited for the apt lock to be
+                # available, it is possible that the lock was acquired by
+                # another process since the check. Since apt doesn't provide
+                # a meaningful error code to check and checking the error
+                # text is fragile and subject to internationalization, we
+                # can instead check the apt lock again. If the apt lock is
+                # still available, given the length of an average apt
+                # transaction, it is extremely unlikely that another process
+                # raced us when we tried to acquire it, so raise the apt
+                # error received. If the lock is unavailable, just keep waiting
+                if self._apt_lock_available():
+                    raise
+                LOG.debug('Another process holds apt lock. Waiting...')
+                time.sleep(1)
+        raise TimeoutError('Could not get apt lock')
+
     def package_command(self, command, args=None, pkgs=None):
+        """Run the given package command.
+
+        On Debian, this will run apt-get (unless APT_GET_COMMAND is set).
+
+        command: The command to run, like "upgrade" or "install"
+        args: Arguments passed to apt itself in addition to
+              any specified in APT_GET_COMMAND
+        pkgs: Apt packages that the command will apply to
+        """
         if pkgs is None:
             pkgs = []
 
@@ -185,11 +264,10 @@ def package_command(self, command, args=None, pkgs=None):
         pkglist = util.expand_package_list('%s=%s', pkgs)
         cmd.extend(pkglist)
 
-        # Allow the output of this to flow outwards (ie not be captured)
-        util.log_time(logfunc=LOG.debug,
-                      msg="apt-%s [%s]" % (command, ' '.join(cmd)),
-                      func=subp.subp,
-                      args=(cmd,), kwargs={'env': e, 'capture': False})
+        self._wait_for_apt_command(
+            short_cmd=command,
+            subp_kwargs={'args': cmd, 'env': e, 'capture': False}
+        )
 
     def update_package_sources(self):
         self._runner.run("update-sources", self.package_command,

tests/unittests/test_distros/test_debian.py

@@ -1,8 +1,16 @@
 # This file is part of cloud-init. See LICENSE file for license information.
+from itertools import count, cycle
+from unittest import mock
 
-from cloudinit import distros
-from cloudinit import util
-from cloudinit.tests.helpers import (FilesystemMockingTestCase, mock)
+import pytest
+
+from cloudinit import distros, util
+from cloudinit.distros.debian import (
+    APT_GET_COMMAND,
+    APT_GET_WRAPPER,
+)
+from cloudinit.tests.helpers import FilesystemMockingTestCase
+from cloudinit import subp
 
 
 @mock.patch("cloudinit.distros.debian.subp.subp")
@@ -98,3 +106,69 @@ def test_falseish_locale_raises_valueerror(self, m_subp):
             m_subp.assert_not_called()
         self.assertEqual(
             'Failed to provide locale value.', str(ctext_m.exception))
+
+
+@mock.patch.dict('os.environ', {}, clear=True)
+@mock.patch("cloudinit.distros.debian.subp.which", return_value=True)
+@mock.patch("cloudinit.distros.debian.subp.subp")
+class TestPackageCommand:
+    distro = distros.fetch("debian")("debian", {}, None)
+
+    @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
+                return_value=True)
+    def test_simple_command(self, m_apt_avail, m_subp, m_which):
+        self.distro.package_command('update')
+        apt_args = [APT_GET_WRAPPER['command']]
+        apt_args.extend(APT_GET_COMMAND)
+        apt_args.append('update')
+        expected_call = {
+            'args': apt_args,
+            'capture': False,
+            'env': {'DEBIAN_FRONTEND': 'noninteractive'},
+        }
+        assert m_subp.call_args == mock.call(**expected_call)
+
+    @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
+                side_effect=[False, False, True])
+    @mock.patch("cloudinit.distros.debian.time.sleep")
+    def test_wait_for_lock(self, m_sleep, m_apt_avail, m_subp, m_which):
+        self.distro._wait_for_apt_command("stub", {"args": "stub2"})
+        assert m_sleep.call_args_list == [mock.call(1), mock.call(1)]
+        assert m_subp.call_args_list == [mock.call(args='stub2')]
+
+    @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
+                return_value=False)
+    @mock.patch("cloudinit.distros.debian.time.sleep")
+    @mock.patch("cloudinit.distros.debian.time.time", side_effect=count())
+    def test_lock_wait_timeout(
+        self, m_time, m_sleep, m_apt_avail, m_subp, m_which
+    ):
+        with pytest.raises(TimeoutError):
+            self.distro._wait_for_apt_command("stub", "stub2", timeout=5)
+        assert m_subp.call_args_list == []
+
+    @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
+                side_effect=cycle([True, False]))
+    @mock.patch("cloudinit.distros.debian.time.sleep")
+    def test_lock_exception_wait(self, m_sleep, m_apt_avail, m_subp, m_which):
+        exception = subp.ProcessExecutionError(
+            exit_code=100, stderr="Could not get apt lock"
+        )
+        m_subp.side_effect = [exception, exception, "return_thing"]
+        ret = self.distro._wait_for_apt_command("stub", {"args": "stub2"})
+        assert ret == "return_thing"
+
+    @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
+                side_effect=cycle([True, False]))
+    @mock.patch("cloudinit.distros.debian.time.sleep")
+    @mock.patch("cloudinit.distros.debian.time.time", side_effect=count())
+    def test_lock_exception_timeout(
+        self, m_time, m_sleep, m_apt_avail, m_subp, m_which
+    ):
+        m_subp.side_effect = subp.ProcessExecutionError(
+            exit_code=100, stderr="Could not get apt lock"
+        )
+        with pytest.raises(TimeoutError):
+            self.distro._wait_for_apt_command(
+                "stub", {"args": "stub2"}, timeout=5
+            )

tests/unittests/test_handler/test_handler_landscape.py

@@ -22,6 +22,10 @@ def setUp(self):
         self.conf = self.tmp_path('client.conf', self.new_root)
         self.default_file = self.tmp_path('default_landscape', self.new_root)
         self.patchUtils(self.new_root)
+        self.add_patch(
+            'cloudinit.distros.ubuntu.Distro.install_packages',
+            'm_install_packages'
+        )
 
     def test_handler_skips_empty_landscape_cloudconfig(self):
         """Empty landscape cloud-config section does no work."""