StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1118 from mwhudson/move-command_runner 

refactor curtin invocation some more
This commit is contained in:
Michael Hudson-Doyle 2021-11-10 09:21:09 +13:00 committed by GitHub
parent 6542f8629f 85f0c62ae4
commit 71b1261731
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 280 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
subiquity/models/subiquity.py
View File

@ -325,7 +325,7 @@ class SubiquityModel:
with open('/etc/machine-id') as fp:
return fp.read()
def render(self, syslog_identifier):
def render(self):
# Until https://bugs.launchpad.net/curtin/+bug/1876984 gets
# fixed, the only way to get curtin to leave the network
# config entirely alone is to omit the 'network' stage.
@ -356,8 +356,6 @@ class SubiquityModel:
'/var/log/installer/curtin-install.log',
},
'verbosity': 3,
'pollinate': {
'user_agent': {
'subiquity': "%s_%s" % (os.environ.get("SNAP_VERSION",
@ -367,13 +365,6 @@ class SubiquityModel:
},
},
'reporting': {
'subiquity': {
'type': 'journald',
'identifier': syslog_identifier,
},
},
'write_files': {
'etc_machine_id': {
'path': 'etc/machine-id',

16
subiquity/models/tests/test_subiquity.py
View File

@ -116,7 +116,7 @@ class TestSubiquityModel(unittest.TestCase):
model = self.make_model()
proxy_val = 'http://my-proxy'
model.proxy.proxy = proxy_val
config = model.render('ident')
config = model.render()
self.assertConfigHasVal(config, 'proxy.http_proxy', proxy_val)
self.assertConfigHasVal(config, 'proxy.https_proxy', proxy_val)
self.assertConfigHasVal(config, 'apt.http_proxy', proxy_val)
@ -129,7 +129,7 @@ class TestSubiquityModel(unittest.TestCase):
def test_proxy_notset(self):
model = self.make_model()
config = model.render('ident')
config = model.render()
self.assertConfigDoesNotHaveVal(config, 'proxy.http_proxy')
self.assertConfigDoesNotHaveVal(config, 'proxy.https_proxy')
self.assertConfigDoesNotHaveVal(config, 'apt.http_proxy')
@ -142,7 +142,7 @@ class TestSubiquityModel(unittest.TestCase):
def test_keyboard(self):
model = self.make_model()
config = model.render('ident')
config = model.render()
self.assertConfigWritesFile(config, 'etc/default/keyboard')
def test_writes_machine_id_media_info(self):
@ -150,18 +150,18 @@ class TestSubiquityModel(unittest.TestCase):
model_proxy = self.make_model()
model_proxy.proxy.proxy = 'http://something'
for model in model_no_proxy, model_proxy:
config = model.render('ident')
config = model.render()
self.assertConfigWritesFile(config, 'etc/machine-id')
self.assertConfigWritesFile(config, 'var/log/installer/media-info')
def test_storage_version(self):
model = self.make_model()
config = model.render('ident')
config = model.render()
self.assertConfigHasVal(config, 'storage.version', 1)
def test_write_netplan(self):
model = self.make_model()
config = model.render('ident')
config = model.render()
netplan_content = None
for fspec in config['write_files'].values():
if fspec['path'].startswith('etc/netplan'):
@ -174,14 +174,14 @@ class TestSubiquityModel(unittest.TestCase):
def test_has_sources(self):
model = self.make_model()
config = model.render('ident')
config = model.render()
self.assertIn('sources', config)
def test_mirror(self):
model = self.make_model()
mirror_val = 'http://my-mirror'
model.mirror.set_mirror(mirror_val)
config = model.render('ident')
config = model.render()
from curtin.commands.apt_config import get_mirror
try:
from curtin.distro import get_architecture

207
subiquity/server/controllers/install.py
View File

@ -13,15 +13,11 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import datetime
import json
import logging
import os
import re
import shutil
import subprocess
import sys
from curtin.commands.install import (
ERROR_TARFILE,
@ -34,21 +30,23 @@ import yaml
from subiquitycore.async_helpers import (
run_in_thread,
)
from subiquitycore.context import Status, with_context
from subiquitycore.utils import (
astart_command,
)
from subiquitycore.context import with_context
from subiquity.common.errorreport import ErrorReportKind
from subiquity.server.controller import (
SubiquityController,
)
from subiquity.common.types import (
ApplicationState,
)
from subiquity.journald import (
journald_listen,
)
from subiquity.server.curtin import (
run_curtin_command,
start_curtin_command,
)
from subiquity.server.controller import (
SubiquityController,
)
log = logging.getLogger("subiquity.server.controllers.install")
@ -72,146 +70,15 @@ class TracebackExtractor:
self.traceback.append(line)
class LoggedCommandRunner:
def __init__(self, ident):
self.ident = ident
async def start(self, cmd):
return await astart_command([
'systemd-cat', '--level-prefix=false', '--identifier='+self.ident,
] + cmd)
async def run(self, cmd):
proc = await self.start(cmd)
await proc.communicate()
if proc.returncode != 0:
raise subprocess.CalledProcessError(proc.returncode, cmd)
else:
return subprocess.CompletedProcess(cmd, proc.returncode)
class DryRunCommandRunner(LoggedCommandRunner):
def __init__(self, ident, delay):
super().__init__(ident)
self.delay = delay
async def start(self, cmd):
if 'scripts/replay-curtin-log.py' in cmd:
delay = 0
else:
cmd = ['echo', 'not running:'] + cmd
if 'unattended-upgrades' in cmd:
delay = 3*self.delay
else:
delay = self.delay
proc = await super().start(cmd)
await asyncio.sleep(delay)
return proc
class CurtinCommandRunner:
def __init__(self, runner, event_syslog_id, config_location):
self.runner = runner
self.event_syslog_id = event_syslog_id
self.config_location = config_location
self._event_contexts = {}
journald_listen(
asyncio.get_event_loop(), [event_syslog_id], self._event)
def _event(self, event):
e = {
"EVENT_TYPE": "???",
"MESSAGE": "???",
"NAME": "???",
"RESULT": "???",
}
prefix = "CURTIN_"
for k, v in event.items():
if k.startswith(prefix):
e[k[len(prefix):]] = v
event_type = e["EVENT_TYPE"]
if event_type == 'start':
def p(name):
parts = name.split('/')
for i in range(len(parts), -1, -1):
yield '/'.join(parts[:i]), '/'.join(parts[i:])
curtin_ctx = None
for pre, post in p(e["NAME"]):
if pre in self._event_contexts:
parent = self._event_contexts[pre]
curtin_ctx = parent.child(post, e["MESSAGE"])
self._event_contexts[e["NAME"]] = curtin_ctx
break
if curtin_ctx:
curtin_ctx.enter()
if event_type == 'finish':
status = getattr(Status, e["RESULT"], Status.WARN)
curtin_ctx = self._event_contexts.pop(e["NAME"], None)
if curtin_ctx is not None:
curtin_ctx.exit(result=status)
def make_command(self, command, *args, **conf):
cmd = [
sys.executable, '-m', 'curtin', '--showtrace',
'-c', self.config_location,
]
for k, v in conf.items():
cmd.extend(['--set', 'json:' + k + '=' + json.dumps(v)])
cmd.append(command)
cmd.extend(args)
return cmd
async def run(self, context, command, *args, **conf):
self._event_contexts[''] = context
await self.runner.run(self.make_command(command, *args, **conf))
waited = 0.0
while len(self._event_contexts) > 1 and waited < 5.0:
await asyncio.sleep(0.1)
waited += 0.1
log.debug("waited %s seconds for events to drain", waited)
self._event_contexts.pop('', None)
class DryRunCurtinCommandRunner(CurtinCommandRunner):
event_file = 'examples/curtin-events.json'
def make_command(self, command, *args, **conf):
if command == 'install':
return [
sys.executable, "scripts/replay-curtin-log.py",
self.event_file, self.event_syslog_id,
'.subiquity' + INSTALL_LOG,
]
else:
return super().make_command(command, *args, **conf)
class FailingDryRunCurtinCommandRunner(DryRunCurtinCommandRunner):
event_file = 'examples/curtin-events-fail.json'
class InstallController(SubiquityController):
def __init__(self, app):
super().__init__(app)
self.model = app.base_model
self.unattended_upgrades_proc = None
self.unattended_upgrades_cmd = None
self.unattended_upgrades_ctx = None
self._event_syslog_id = 'curtin_event.%s' % (os.getpid(),)
self.tb_extractor = TracebackExtractor()
if self.app.opts.dry_run:
self.command_runner = DryRunCommandRunner(
self.app.log_syslog_id, 2/self.app.scale_factor)
else:
self.command_runner = LoggedCommandRunner(self.app.log_syslog_id)
self.curtin_runner = None
def stop_uu(self):
if self.app.state == ApplicationState.UU_RUNNING:
@ -229,8 +96,8 @@ class InstallController(SubiquityController):
def log_event(self, event):
self.tb_extractor.feed(event['MESSAGE'])
def make_curtin_command_runner(self):
config = self.model.render(syslog_identifier=self._event_syslog_id)
def write_config(self):
config = self.model.render()
config_location = '/var/log/installer/subiquity-curtin-install.conf'
log_location = INSTALL_LOG
if self.app.opts.dry_run:
@ -246,26 +113,19 @@ class InstallController(SubiquityController):
self.app.note_file_for_apport("CurtinConfig", config_location)
self.app.note_file_for_apport("CurtinErrors", ERROR_TARFILE)
self.app.note_file_for_apport("CurtinLog", log_location)
if self.app.opts.dry_run:
if 'install-fail' in self.app.debug_flags:
cls = FailingDryRunCurtinCommandRunner
else:
cls = DryRunCurtinCommandRunner
else:
cls = CurtinCommandRunner
self.curtin_runner = cls(
self.command_runner, self._event_syslog_id, config_location)
return config_location
@with_context(description="umounting /target dir")
async def unmount_target(self, *, context, target):
await self.curtin_runner.run(context, 'unmount', '-t', target)
await run_curtin_command(self.app, context, 'unmount', '-t', target)
if not self.app.opts.dry_run:
shutil.rmtree(target)
@with_context(
description="installing system", level="INFO", childlevel="DEBUG")
async def curtin_install(self, *, context):
await self.curtin_runner.run(context, 'install')
async def curtin_install(self, *, context, config_location):
await run_curtin_command(
self.app, context, 'install', config=config_location)
@with_context()
async def install(self, *, context):
@ -287,13 +147,14 @@ class InstallController(SubiquityController):
self.app.update_state(ApplicationState.RUNNING)
self.make_curtin_command_runner()
config_location = self.write_config()
if os.path.exists(self.model.target):
await self.unmount_target(
context=context, target=self.model.target)
await self.curtin_install(context=context)
await self.curtin_install(
context=context, config_location=config_location)
self.app.update_state(ApplicationState.POST_WAIT)
@ -344,17 +205,18 @@ class InstallController(SubiquityController):
name="install_{package}",
description="installing {package}")
async def install_package(self, *, context, package):
await self.curtin_runner.run(context, 'system-install', '--', package)
await run_curtin_command(
self.app, context, 'system-install', '--', package)
@with_context(description="restoring apt configuration")
async def restore_apt_config(self, context):
await self.command_runner.run(["umount", self.tpath('etc/apt')])
await self.app.command_runner.run(["umount", self.tpath('etc/apt')])
if self.model.network.has_network:
await self.curtin_runner.run(
context, "in-target", "-t", self.tpath(),
await run_curtin_command(
self.app, context, "in-target", "-t", self.tpath(),
"--", "apt-get", "update")
else:
await self.command_runner.run(
await self.app.command_runner.run(
["umount", self.tpath('var/lib/apt/lists')])
@with_context(description="downloading and installing {policy} updates")
@ -374,27 +236,26 @@ class InstallController(SubiquityController):
apt_conf.write(apt_conf_contents)
apt_conf.close()
self.unattended_upgrades_ctx = context
self.unattended_upgrades_proc = await self.command_runner.start(
self.curtin_runner.make_command(
"in-target", "-t", self.tpath(),
"--", "unattended-upgrades", "-v"))
await self.unattended_upgrades_proc.communicate()
self.unattended_upgrades_proc = None
self.unattended_upgrades_cmd = await start_curtin_command(
self.app, context, "in-target", "-t", self.tpath(),
"--", "unattended-upgrades", "-v")
await self.unattended_upgrades_cmd.wait()
self.unattended_upgrades_cmd = None
self.unattended_upgrades_ctx = None
async def stop_unattended_upgrades(self):
with self.unattended_upgrades_ctx.parent.child(
"stop_unattended_upgrades",
"cancelling update"):
await self.command_runner.run([
await self.app.command_runner.run([
'chroot', self.tpath(),
'/usr/share/unattended-upgrades/'
'unattended-upgrade-shutdown',
'--stop-only',
])
if self.app.opts.dry_run and \
self.unattended_upgrades_proc is not None:
self.unattended_upgrades_proc.terminate()
self.unattended_upgrades_cmd is not None:
self.unattended_upgrades_cmd.proc.terminate()
uu_apt_conf = b"""\

164
subiquity/server/curtin.py Normal file
View File

@ -0,0 +1,164 @@
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import json
import logging
import os
import sys
from curtin.commands.install import (
INSTALL_LOG,
)
from subiquitycore.context import Status
from subiquity.journald import (
journald_listen,
)
log = logging.getLogger('subiquity.server.curtin')
class _CurtinCommand:
_count = 0
def __init__(self, runner, command, *args, config=None):
self.runner = runner
self._event_contexts = {}
_CurtinCommand._count += 1
self._event_syslog_id = 'curtin_event.%s.%s' % (
os.getpid(), _CurtinCommand._count)
self._fd = None
self.proc = None
self._cmd = self.make_command(command, *args, config=config)
def _event(self, event):
e = {
"EVENT_TYPE": "???",
"MESSAGE": "???",
"NAME": "???",
"RESULT": "???",
}
prefix = "CURTIN_"
for k, v in event.items():
if k.startswith(prefix):
e[k[len(prefix):]] = v
event_type = e["EVENT_TYPE"]
if event_type == 'start':
def p(name):
parts = name.split('/')
for i in range(len(parts), -1, -1):
yield '/'.join(parts[:i]), '/'.join(parts[i:])
curtin_ctx = None
for pre, post in p(e["NAME"]):
if pre in self._event_contexts:
parent = self._event_contexts[pre]
curtin_ctx = parent.child(post, e["MESSAGE"])
self._event_contexts[e["NAME"]] = curtin_ctx
break
if curtin_ctx:
curtin_ctx.enter()
if event_type == 'finish':
status = getattr(Status, e["RESULT"], Status.WARN)
curtin_ctx = self._event_contexts.pop(e["NAME"], None)
if curtin_ctx is not None:
curtin_ctx.exit(result=status)
def make_command(self, command, *args, config=None):
reporting_conf = {
'subiquity': {
'type': 'journald',
'identifier': self._event_syslog_id,
},
}
cmd = [
sys.executable, '-m', 'curtin', '--showtrace', '-vvv',
'--set', 'json:reporting=' + json.dumps(reporting_conf),
]
if config is not None:
cmd.extend([
'-c', config,
])
cmd.append(command)
cmd.extend(args)
return cmd
async def start(self, context):
self._fd = journald_listen(
asyncio.get_event_loop(), [self._event_syslog_id], self._event)
# Yield to the event loop before starting curtin to avoid missing the
# first couple of events.
await asyncio.sleep(0)
self._event_contexts[''] = context
self.proc = await self.runner.start(self._cmd)
async def wait(self):
await self.runner.wait(self.proc)
waited = 0.0
while len(self._event_contexts) > 1 and waited < 5.0:
await asyncio.sleep(0.1)
waited += 0.1
log.debug("waited %s seconds for events to drain", waited)
self._event_contexts.pop('', None)
asyncio.get_event_loop().remove_reader(self._fd)
async def run(self, context):
await self.start(context)
await self.wait()
class _DryRunCurtinCommand(_CurtinCommand):
event_file = 'examples/curtin-events.json'
def make_command(self, command, *args, config=None):
if command == 'install':
return [
sys.executable,
"scripts/replay-curtin-log.py",
self.event_file,
self._event_syslog_id,
'.subiquity' + INSTALL_LOG,
]
else:
return super().make_command(command, *args, config=config)
class _FailingDryRunCurtinCommand(_DryRunCurtinCommand):
event_file = 'examples/curtin-events-fail.json'
async def start_curtin_command(app, context, command, *args, config=None):
if app.opts.dry_run:
if 'install-fail' in app.debug_flags:
cls = _FailingDryRunCurtinCommand
else:
cls = _DryRunCurtinCommand
else:
cls = _CurtinCommand
curtin_cmd = cls(app.command_runner, command, *args, config=config)
await curtin_cmd.start(context)
return curtin_cmd
async def run_curtin_command(app, context, command, *args, config=None):
cmd = await start_curtin_command(
app, context, command, *args, config=config)
await cmd.wait()

71
subiquity/server/runner.py Normal file
View File

@ -0,0 +1,71 @@
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import subprocess
from subiquitycore.utils import astart_command
class LoggedCommandRunner:
def __init__(self, ident):
self.ident = ident
async def start(self, cmd):
proc = await astart_command([
'systemd-cat', '--level-prefix=false', '--identifier='+self.ident,
] + cmd)
proc.args = cmd
return proc
async def wait(self, proc):
await proc.communicate()
if proc.returncode != 0:
raise subprocess.CalledProcessError(proc.returncode, proc.args)
else:
return subprocess.CompletedProcess(proc.args, proc.returncode)
async def run(self, cmd):
proc = await self.start(cmd)
return await self.wait(proc)
class DryRunCommandRunner(LoggedCommandRunner):
def __init__(self, ident, delay):
super().__init__(ident)
self.delay = delay
async def start(self, cmd):
if 'scripts/replay-curtin-log.py' in cmd:
delay = 0
else:
cmd = ['echo', 'not running:'] + cmd
if 'unattended-upgrades' in cmd:
delay = 3*self.delay
else:
delay = self.delay
proc = await super().start(cmd)
await asyncio.sleep(delay)
return proc
def get_command_runner(app):
if app.opts.dry_run:
return DryRunCommandRunner(
app.log_syslog_id, 2/app.scale_factor)
else:
return LoggedCommandRunner(app.log_syslog_id)

2
subiquity/server/server.py
View File

@ -68,6 +68,7 @@ from subiquity.models.subiquity import (
from subiquity.server.controller import SubiquityController
from subiquity.server.geoip import GeoIP
from subiquity.server.errors import ErrorController
from subiquity.server.runner import get_command_runner
from subiquity.server.types import InstallerChannels
from subiquitycore.snapd import (
AsyncSnapd,
@ -269,6 +270,7 @@ class SubiquityServer(Application):
self.echo_syslog_id = 'subiquity_echo.{}'.format(os.getpid())
self.event_syslog_id = 'subiquity_event.{}'.format(os.getpid())
self.log_syslog_id = 'subiquity_log.{}'.format(os.getpid())
self.command_runner = get_command_runner(self)
self.error_reporter = ErrorReporter(
self.context.child("ErrorReporter"), self.opts.dry_run, self.root)