4.3.1. System Configuration#

4.3.1.1. Base#

class simbricks.orchestration.system.base.System(name: str = 'def-sys-name')#

Defines System configuration of the whole simulation

get_comp(ident: int) → Component#
get_inf(ident: int) → Interface#
get_chan(ident: int) → Channel#
get_app(ident: int) → app.Application#
static set_latencies(channels: list[Channel], latency: TimeInterval | str) → None#
latencies(latency: TimeInterval | str, channel_type: Any | None = None) → None#
toJSON() → dict#
classmethod fromJSON(json_obj: dict, enforce_dummies: bool = False) → Self#
id() → int#
class simbricks.orchestration.system.base.Component(s: System)#

Defines a component that is a part of the System Configuration.

Components could e.g. be hosts, NICs, switches.

interfaces() → list[Interface]#
add_if(interface: Interface) → None#
channels() → list[Channel]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
class simbricks.orchestration.system.base.DummyComponent(s: System)#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_if(interface: Interface) → None#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#
class simbricks.orchestration.system.base.GenericComponent(s: System, name: str | None = None)#

A component without any simulator-specific semantics.

Meant to be paired with simbricks.orchestration.simulation.base.GenericSimulator to model a simulator that SimBricks has no dedicated Python class for. Interfaces are not created for you, add them explicitly:

comp = GenericComponent(system, name="my-dev")
eth_if = EthInterface(comp)
comp.add_if(eth_if)

or let one of the helpers in simbricks.orchestration.helpers.system do it.

This class carries no state beyond what Component already serializes, hence it needs no toJSON/fromJSON of its own.

add_if(interface: Interface) → None#
channels() → list[Channel]#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#
class simbricks.orchestration.system.base.Interface(c: Component)#

Specifies a single Interface of a Component.

A host component could e.g. have multiple PCI Interfaces.

property component: Component#
is_connected() → bool#
disconnect() → None#
connect(c: Channel) → None#
find_peer() → Interface#
get_chan_raise() → Channel#
get_opposing_interface() → Interface#
T = ~T#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

id() → int#
class simbricks.orchestration.system.base.DummyInterface(c: Component)#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

T = ~T#
property component: Component#
connect(c: Channel) → None#
disconnect() → None#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
find_peer() → Interface#
get_chan_raise() → Channel#
get_opposing_interface() → Interface#
id() → int#
is_connected() → bool#
toJSON() → dict#
channel: Channel | None#
class simbricks.orchestration.system.base.Channel(a: Interface, b: Interface)#
interfaces() → tuple[Interface, Interface]#
disconnect()#
get_opposing_interface(interface: Interface) → Interface#
property latency: TimeInterval#

Link latency/propagation delay of this channel.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#

4.3.1.2. Eth#

class simbricks.orchestration.system.eth.EthInterface(c: Component)#
connect(c: Channel) → None#
T = ~T#
property component: Component#
disconnect() → None#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
find_peer() → Interface#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

get_chan_raise() → Channel#
get_opposing_interface() → Interface#
id() → int#
is_connected() → bool#
toJSON() → dict#
channel: Channel | None#
class simbricks.orchestration.system.eth.EthChannel(a: EthInterface, b: EthInterface)#
disconnect()#
classmethod fromJSON(system: System, json_obj: dict) → Self#
get_opposing_interface(interface: Interface) → Interface#
id() → int#
interfaces() → tuple[Interface, Interface]#
property latency: TimeInterval#

Link latency/propagation delay of this channel.

toJSON() → dict#
class simbricks.orchestration.system.eth.EthSimpleNIC(s: System)#
add_ipv4(ip: str) → None#
add_if(interface: EthInterface) → None#
connect_eth_peer_if(peer_i: EthInterface) → EthChannel#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
class simbricks.orchestration.system.eth.BaseEthNetComponent(s: System)#
add_if(i: EthInterface) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
connect_eth_peer_if(peer_i: EthInterface) → EthChannel#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
class simbricks.orchestration.system.eth.EthWire(s: System)#
add_if(i: EthInterface) → None#
channels() → list[Channel]#
connect_eth_peer_if(peer_i: EthInterface) → EthChannel#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#
class simbricks.orchestration.system.eth.EthSwitch(s: System)#
add_if(i: EthInterface) → None#
channels() → list[Channel]#
connect_eth_peer_if(peer_i: EthInterface) → EthChannel#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#

4.3.1.3. PCIe#

class simbricks.orchestration.system.pcie.PCIeHostInterface(c: Component)#
T = ~T#
property component: Component#
connect(c: Channel) → None#
disconnect() → None#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
find_peer() → Interface#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

get_chan_raise() → Channel#
get_opposing_interface() → Interface#
id() → int#
is_connected() → bool#
toJSON() → dict#
channel: Channel | None#
class simbricks.orchestration.system.pcie.PCIeDeviceInterface(c: Component)#
connect(c: Channel) → None#
T = ~T#
property component: Component#
disconnect() → None#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
find_peer() → Interface#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

get_chan_raise() → Channel#
get_opposing_interface() → Interface#
id() → int#
is_connected() → bool#
toJSON() → dict#
channel: Channel | None#
class simbricks.orchestration.system.pcie.PCIeChannel(host: PCIeHostInterface, dev: PCIeDeviceInterface)#
host_if() → PCIeHostInterface#
dev_if() → PCIeDeviceInterface#
disconnect()#
classmethod fromJSON(system: System, json_obj: dict) → Self#
get_opposing_interface(interface: Interface) → Interface#
id() → int#
interfaces() → tuple[Interface, Interface]#
property latency: TimeInterval#

Link latency/propagation delay of this channel.

toJSON() → dict#
class simbricks.orchestration.system.pcie.PCIeSimpleDevice(s: System)#
add_if(interface: PCIeDeviceInterface) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
class simbricks.orchestration.system.pcie.NVMeSSD(s: System)#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_if(interface: PCIeDeviceInterface) → None#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#

4.3.1.4. Mem#

class simbricks.orchestration.system.mem.MemHostInterface(c: Component)#
T = ~T#
property component: Component#
connect(c: Channel) → None#
disconnect() → None#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
find_peer() → Interface#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

get_chan_raise() → Channel#
get_opposing_interface() → Interface#
id() → int#
is_connected() → bool#
toJSON() → dict#
channel: Channel | None#
class simbricks.orchestration.system.mem.MemDeviceInterface(c: Component)#
connect(c: Channel) → None#
T = ~T#
property component: Component#
disconnect() → None#
static filter_by_type(interfaces: list[Interface], ty: type[T]) → list[T]#
find_peer() → Interface#
classmethod fromJSON(system: System, json_obj: dict) → Self#

Deserializes an Interface from its JSON representation.

Note: This function will not set the component which is done by the component base classes ‘fromJSON’ method.

get_chan_raise() → Channel#
get_opposing_interface() → Interface#
id() → int#
is_connected() → bool#
toJSON() → dict#
channel: Channel | None#
class simbricks.orchestration.system.mem.MemChannel(host: MemHostInterface, dev: MemDeviceInterface)#
host_if() → MemHostInterface#
dev_if() → MemDeviceInterface#
disconnect()#
classmethod fromJSON(system: System, json_obj: dict) → Self#
get_opposing_interface(interface: Interface) → Interface#
id() → int#
interfaces() → tuple[Interface, Interface]#
property latency: TimeInterval#

Link latency/propagation delay of this channel.

toJSON() → dict#
class simbricks.orchestration.system.mem.MemSimpleDevice(s: System)#
add_if(interface: MemDeviceInterface) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
class simbricks.orchestration.system.mem.MemInterconnect(s: System)#
add_if(interface: MemDeviceInterface | MemHostInterface) → None#
connect_host(peer_if: MemHostInterface) → MemChannel#
connect_device(peer_if: MemDeviceInterface) → MemChannel#
add_route(dev: MemHostInterface, vaddr: int, len: int, paddr: int = 0)#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
class simbricks.orchestration.system.mem.MemTerminal(s: System)#
add_if(interface: MemDeviceInterface) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#

4.3.1.5. NIC#

class simbricks.orchestration.system.nic.SimplePCIeNIC(s: System)#
add_if(interface: EthInterface | PCIeDeviceInterface) → None#
add_ipv4(ip: str) → None#
channels() → list[Channel]#
connect_eth_peer_if(peer_i: EthInterface) → EthChannel#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#
class simbricks.orchestration.system.nic.VirtIONic(s: System)#
add_if(interface: EthInterface | PCIeDeviceInterface) → None#
add_ipv4(ip: str) → None#
channels() → list[Channel]#
connect_eth_peer_if(peer_i: EthInterface) → EthChannel#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
toJSON() → dict#

4.3.1.6. Host#

class simbricks.orchestration.system.host.base.Host(s: System)#
add_app(a: Application) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
connect_pcie_dev(dev: PCIeSimpleDevice) → PCIeChannel#
add_if(interface: Interface) → None#
channels() → list[Channel]#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: inst_base.Instantiation) → None#
class simbricks.orchestration.system.host.base.FullSystemHost(s: System)#
add_disk(disk: DiskImage) → None#
async prepare(inst: instantiation.Instantiation) → None#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_app(a: Application) → None#
add_if(interface: Interface) → None#
channels() → list[Channel]#
connect_pcie_dev(dev: PCIeSimpleDevice) → PCIeChannel#
id() → int#
interfaces() → list[Interface]#
class simbricks.orchestration.system.host.base.BaseLinuxHost(s: System)#
add_app(a: BaseLinuxApplication) → None#
run_cmds(inst: instantiation.Instantiation) → list[str]#

Commands to run on node.

cleanup_cmds(inst: instantiation.Instantiation) → list[str]#

Commands to run to cleanup node.

add_config_file(config_file: ConfigFile)#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

config_files(inst: instantiation.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

prepare_pre_cp(inst: instantiation.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

prepare_post_cp(inst: instantiation.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

config_str(inst: instantiation.Instantiation) → str#
toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_disk(disk: DiskImage) → None#
add_if(interface: Interface) → None#
channels() → list[Channel]#
connect_pcie_dev(dev: PCIeSimpleDevice) → PCIeChannel#
id() → int#
interfaces() → list[Interface]#
async prepare(inst: instantiation.Instantiation) → None#
class simbricks.orchestration.system.host.base.LinuxHost(sys)#
cleanup_cmds(inst: instantiation.Instantiation) → list[str]#

Commands to run to cleanup node.

prepare_pre_cp(inst: instantiation.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

prepare_post_cp(inst) → list[str]#

Commands to run to prepare node after checkpoint restore.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_app(a: BaseLinuxApplication) → None#
add_config_file(config_file: ConfigFile)#
add_disk(disk: DiskImage) → None#
add_if(interface: Interface) → None#
channels() → list[Channel]#
config_files(inst: instantiation.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

config_str(inst: instantiation.Instantiation) → str#
connect_pcie_dev(dev: PCIeSimpleDevice) → PCIeChannel#
id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

interfaces() → list[Interface]#
async prepare(inst: instantiation.Instantiation) → None#
run_cmds(inst: instantiation.Instantiation) → list[str]#

Commands to run on node.

class simbricks.orchestration.system.host.base.NVMeLinuxHost(sys: System)#
add_app(a: BaseLinuxApplication) → None#
add_config_file(config_file: ConfigFile)#
add_disk(disk: DiskImage) → None#
add_if(interface: Interface) → None#
channels() → list[Channel]#
cleanup_cmds(inst: instantiation.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: instantiation.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

config_str(inst: instantiation.Instantiation) → str#
connect_pcie_dev(dev: PCIeSimpleDevice) → PCIeChannel#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

interfaces() → list[Interface]#
async prepare(inst: instantiation.Instantiation) → None#
prepare_post_cp(inst) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: instantiation.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

run_cmds(inst: instantiation.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#

4.3.1.7. Applications#

class simbricks.orchestration.system.host.app.Application(h: sys_host.Host)#
toJSON() → dict#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
class simbricks.orchestration.system.host.app.BaseLinuxApplication(h: sys_host.LinuxHost)#
abstractmethod run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

add_config_file(config_file: disk_images.ConfigFile)#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
id() → int#
class simbricks.orchestration.system.host.app.GenericRawCommandApplication(h: sys_host.LinuxHost, commands: list[str] = [])#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.NVMEFsTest(h: sys_host.LinuxHost)#
classmethod fromJSON(system: System, json_obj: dict) → Self#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

toJSON() → dict#
class simbricks.orchestration.system.host.app.PingClient(h: sys_host.LinuxHost, server_ip: str = '192.168.64.1')#
run_cmds(inst: inst_base.Instantiation) → tp.List[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.Sleep(h: sys_host.LinuxHost, delay: float = 10, infinite: bool = False)#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.NetperfServer(h: sys_host.LinuxHost)#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

toJSON() → dict#
class simbricks.orchestration.system.host.app.NetperfClient(h: sys_host.LinuxHost, server_ip: str = '192.168.64.1')#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.IperfTCPServer(h: sys_host.LinuxHost)#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.IperfUDPServer(h: sys_host.LinuxHost)#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.IperfTCPClient(h: sys_host.LinuxHost, server_ip: str = '10.0.0.1', procs: int = 1)#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

class simbricks.orchestration.system.host.app.IperfUDPClient(h: sys_host.LinuxHost, server_ip: str = '10.0.0.1', rate: str = '150m')#
run_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run on node.

toJSON() → dict#
classmethod fromJSON(system: System, json_obj: dict) → Self#
add_config_file(config_file: disk_images.ConfigFile)#
cleanup_cmds(inst: inst_base.Instantiation) → list[str]#

Commands to run to cleanup node.

config_files(inst: inst_base.Instantiation) → list[disk_images.ConfigFile]#

Additional files to put inside the node, which are mounted under /tmp/guest/.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prepare_post_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node after checkpoint restore.

prepare_pre_cp(inst: inst_base.Instantiation) → list[str]#

Commands to run to prepare node before checkpointing.

4.3.1.8. Disk Images#

class simbricks.orchestration.system.disk_images.BootArtifact(*values)#

A boot file inside a disk image that a simulator needs handed to it separately.

Simulators that boot a kernel themselves cannot read it out of the image, so the image hands it over: QEMU takes the compressed kernel, gem5 the uncompressed ELF. The value doubles as the file name.

VMLINUZ = 'vmlinuz'#
INITRD = 'initrd'#
VMLINUX = 'vmlinux'#
simbricks.orchestration.system.disk_images.hash_strings(parts: Iterable[str]) → str#

Hash of a list of strings, with the list structure part of what is hashed.

simbricks.orchestration.system.disk_images.hash_file(handle: IO) → str#
simbricks.orchestration.system.disk_images.guestfs_env() → list[str]#

Force the direct appliance backend unless one is already chosen.

‘direct’ needs no libvirt session. env(1) because commands run without a shell, inheriting the executor’s environment.

simbricks.orchestration.system.disk_images.version_key(value: str) → list[tuple[int, object]]#

Sort key comparing digit runs numerically, so -100 beats -99.

class simbricks.orchestration.system.disk_images.DiskImage(system: sys_base.System)#
abstractmethod available_formats() → list[str]#
abstractmethod path(inst: inst_base.Instantiation, format: str) → str#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

static assert_is_file(path: str) → None#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files from inside this image, as paths keyed by kind. Only successfully retrieved boot artifacts are returned. Simulators later check if boot artifact is actually available.

Simulators ask for every kind at once, so an image that has to extract them only pays for that once.

find_format(host: sim_host.HostSim) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
id() → int#
class simbricks.orchestration.system.disk_images.DummyDiskImage(system: sys_base.System)#

Stand-in for a disk image whose class could not be imported.

Usually the package providing it is not installed here, so it keeps what it stood in for and says so when something needs the image.

orig_type: str | None = None#
orig_module: str | None = None#
available_formats() → list[str]#
path(inst: inst_base.Instantiation, format: str) → str#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files from inside this image, as paths keyed by kind. Only successfully retrieved boot artifacts are returned. Simulators later check if boot artifact is actually available.

Simulators ask for every kind at once, so an image that has to extract them only pays for that once.

classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
class simbricks.orchestration.system.disk_images.ExternalDiskImage(system: sys_base.System, path: StrPath, boot_dir: str | None = None)#
available_formats() → list[str]#
path(inst: inst_base.Instantiation, format: str) → str#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files from inside this image, as paths keyed by kind. Only successfully retrieved boot artifacts are returned. Simulators later check if boot artifact is actually available.

Simulators ask for every kind at once, so an image that has to extract them only pays for that once.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
class simbricks.orchestration.system.disk_images.ExternalDiskImageArtifact(system: sys_base.System, path: StrPath, boot_dir: str | None = None)#

A disk image on the submitting machine, shipped to the runner with the run.

For a base image small enough to travel inside the run event, which is how input artifacts are carried. Anything sizeable belongs in an HttpDiskImage or in the runner’s global input directory.

input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

path(inst: inst_base.Instantiation, format: str) → str#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files from inside this image, as paths keyed by kind. Only successfully retrieved boot artifacts are returned. Simulators later check if boot artifact is actually available.

Simulators ask for every kind at once, so an image that has to extract them only pays for that once.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
class simbricks.orchestration.system.disk_images.DynamicDiskImage(system: sys_base.System)#
virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

path(inst: inst_base.Instantiation, format: str) → str#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
abstractmethod available_formats() → list[str]#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
class simbricks.orchestration.system.disk_images.FileDiskImage(system: sys_base.System, path: StrPath)#

An image already on the runner, at a path the user gives.

Given the same way as an ExternalDiskImage and treated the opposite way: that one is a black box, handed to the simulator as it is, while this one is a DynamicDiskImage, so its boot artifacts are extracted from the image and cached like any other. Reach for ExternalDiskImage when an image should be left alone, and for this when it should not.

available_formats() → list[str]#
path(inst: inst_base.Instantiation, format: str) → str#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.FileDiskImageArtifact(system: sys_base.System, path: StrPath)#

A disk image on the submitting machine, shipped to the runner with the run.

To ExternalDiskImageArtifact what FileDiskImage is to ExternalDiskImage: the same way in, but its boot artifacts are extracted from the image rather than having to be supplied.

input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

path(inst: inst_base.Instantiation, format: str) → str#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.HttpDiskImage(system: sys_base.System, url: str, checksum: str | None = None, format: str = 'qcow2', qemu_img_exec: str = 'qemu-img')#

A base image the runner downloads, rather than one it already has.

The download is cached like a build is, so a run pays for it once per URL and checksum however many runs use it.

available_formats() → list[str]#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.DistroDiskImage(system: sys_base.System, image_name: str, image_version: str, format: str = 'qcow2', qemu_img_exec: str = 'qemu-img')#

An image from the SimBricks image registry, which the runner downloads.

Images are published as packages under <BASE_URL>/<name>/<version>, holding the image itself as <name> and its sha256 next to it as <name>.sha256.

BASE_URL = 'https://disk-images.simbricks.io'#
classmethod package_url(image_name: str, image_version: str) → str#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.Ubuntu2204BaseDiskImage(system: sys_base.System, qemu_img_exec: str = 'qemu-img')#
BASE_URL = 'https://disk-images.simbricks.io'#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
id() → int#
classmethod package_url(image_name: str, image_version: str) → str#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.Ubuntu2404BaseDiskImage(system: sys_base.System, qemu_img_exec: str = 'qemu-img')#
BASE_URL = 'https://disk-images.simbricks.io'#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
id() → int#
classmethod package_url(image_name: str, image_version: str) → str#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.Ubuntu2204CustomKernelDiskImage(system: sys_base.System, qemu_img_exec: str = 'qemu-img')#
BASE_URL = 'https://disk-images.simbricks.io'#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
id() → int#
classmethod package_url(image_name: str, image_version: str) → str#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.Ubuntu2404CustomKernelDiskImage(system: sys_base.System, qemu_img_exec: str = 'qemu-img')#
BASE_URL = 'https://disk-images.simbricks.io'#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
id() → int#
classmethod package_url(image_name: str, image_version: str) → str#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.LinuxConfigDiskImage(system: sys_base.System, host: sys_host.BaseLinuxHost)#
available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

The LinuxConfigDiskImage does not provide any boot artifacts, since it is not a bootable disk image.

toJSON()#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.BaseLinuxHost) → None#
static assert_is_file(path: str) → None#
content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.PackerDiskImage(system: sys_base.System, packer_config_path: StrPath)#
available_formats() → list[str]#
toJSON() → dict#
classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#
add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

class simbricks.orchestration.system.disk_images.ConfigFile(file_name: str)#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

abstractmethod IOHandle(inst: inst_base.Instantiation) → tp.IO#
toJSON() → dict#
classmethod fromJSON(json_obj) → Self#
id() → int#
class simbricks.orchestration.system.disk_images.ConfigFileLocal(file_name: str, path: str | PathLike[str])#
IOHandle(inst: inst_base.Instantiation) → tp.IO#
toJSON() → dict#
classmethod fromJSON(json_obj) → Self#
id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

class simbricks.orchestration.system.disk_images.ConfigFileStr(file_name: str, string: str)#
IOHandle(inst: inst_base.Instantiation) → tp.IO#
toJSON() → dict#
classmethod fromJSON(json_obj) → Self#
id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

class simbricks.orchestration.system.disk_images.ConfigFileArtifact(file_name: str, path: str | PathLike[str])#

Local filei, shipped to the runner if necessary as input artifact.

input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

IOHandle(inst: inst_base.Instantiation) → tp.IO#
id() → int#
toJSON() → dict#
classmethod fromJSON(json_obj) → Self#

4.3.1.9. Image Layers#

A disk image built as a base image plus an ordered list of changes.

Layers are declarative rather than shell, so the same list can be lowered to virt-customize flags, packer scripts, or Dockerfile instructions. None of the three operations is distro-specific: installing a package is just a command. File payloads are ConfigFiles, which already handle getting a local file to the runner.

class simbricks.orchestration.system.image_layers.ImageLayer#

One step in building an image.

config_files() → list[ConfigFile]#

Files this layer puts into the image, if any.

input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

abstractmethod hash_parts(inst: inst_base.Instantiation) → list[str]#

What this layer contributes to the image’s content hash.

Everything that changes what the layer does to the image, and nothing that does not – an id or a scratch path would make every run unique.

classmethod fromJSON(json_obj: dict) → Self#
id() → int#
toJSON()#
class simbricks.orchestration.system.image_layers.AddFiles(files: list[ConfigFile], dest_dir: str, mode: int | None = None)#

Put files into @dest_dir in the image, optionally chmod’ed to @mode.

config_files() → list[ConfigFile]#

Files this layer puts into the image, if any.

hash_parts(inst: inst_base.Instantiation) → list[str]#

What this layer contributes to the image’s content hash.

Everything that changes what the layer does to the image, and nothing that does not – an id or a scratch path would make every run unique.

toJSON() → dict#
classmethod fromJSON(json_obj: dict) → Self#
id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

class simbricks.orchestration.system.image_layers.RunCommand(cmd: str)#

Run a shell command inside the image.

hash_parts(inst: inst_base.Instantiation) → list[str]#

What this layer contributes to the image’s content hash.

Everything that changes what the layer does to the image, and nothing that does not – an id or a scratch path would make every run unique.

toJSON() → dict#
classmethod fromJSON(json_obj: dict) → Self#
config_files() → list[ConfigFile]#

Files this layer puts into the image, if any.

id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

class simbricks.orchestration.system.image_layers.RunScript(file: ConfigFile)#

Run a script inside the image. Inline or taken from the submitting machine.

config_files() → list[ConfigFile]#

Files this layer puts into the image, if any.

hash_parts(inst: inst_base.Instantiation) → list[str]#

What this layer contributes to the image’s content hash.

Everything that changes what the layer does to the image, and nothing that does not – an id or a scratch path would make every run unique.

toJSON() → dict#
classmethod fromJSON(json_obj: dict) → Self#
id() → int#
input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

simbricks.orchestration.system.image_layers.parse_size(size: str) → int#

Bytes for a size written the way qemu and packer take it, e.g. “32G”.

class simbricks.orchestration.system.image_layers.LayeredDiskImage(system: sys_base.System, base: disk_images.DiskImage)#

A base image plus layers, built when the simulation is prepared.

Subclasses supplying the build live in their own packages, so a runner only installs the tooling for the ways it builds images. Failing to import one is the intended signal that a runner cannot build this image.

disk_size: str | None#

Grow the image to this size, e.g. “32G”, before the layers run. None keeps the base’s size, and a size below it is left alone rather than shrinking the image. Not a layer: the build materializes one image, so this can only happen as it is created.

qemu_img_exec#

What the image, and the chain it may be a delta on, are manipulated with – here and in the backends, which all build qcow2.

add_layer(layer: ImageLayer) → ImageLayer#
run(cmd: str) → ImageLayer#

Run a shell command in the image.

run_script(path: str | PathLike[str]) → ImageLayer#

Run a script taken from the submitting machine.

run_script_str(name: str, script: str) → ImageLayer#

Run a script given inline.

add_file(dest: str, content: str, mode: int | None = None) → ImageLayer#

Write @content to @dest in the image.

copy_in(src: str | PathLike[str], dest: str, mode: int | None = None) → ImageLayer#

Copy a file from the submitting machine to @dest in the image.

input_artifact_files() → list[str]#

Paths of the local files needed, relative to where the script defining them runs.

prefix_hashes(inst: inst_base.Instantiation) → list[str]#

Content hash after each layer, so a build can start from the longest prefix that was cached. The first entry is the base with no layers.

The builder class is in the seed: the same layers on the same base give different images depending on which backend ran them.

content_hash(inst: inst_base.Instantiation) → str#

Identity of what this image holds, for reusing a build across runs.

Two images with the same hash are interchangeable. Raising is the right answer for an image that is built fresh for every run.

abstractmethod async build(inst: inst_base.Instantiation, format: str, base_path: str, layers: list[ImageLayer], out_path: str, overlay: bool = False) → None#

Produce @out_path in @format from @base_path by applying @layers.

With @overlay, @out_path holds only what the layers changed and reads the rest from @base_path, which it must therefore keep referring to.

async convert_image(inst: inst_base.Instantiation, source: str, out: str, format: str) → None#

Write @source, and anything it is a delta on, to @out as one @format image.

async overlay_image(inst: inst_base.Instantiation, backing: str, out: str) → None#

Put an empty image at @out that reads everything from @backing.

async compress_image(inst: inst_base.Instantiation, source: str, out: str, backing: str | None, algorithm: str) → None#

Write @source to @out compressed with @algorithm, holding the same data. With @backing, @out stays a delta on it rather than growing into a whole image.

async rebase_image(inst: inst_base.Instantiation, image: str, backing: str) → None#

Record @backing as where @image reads what it does not hold itself. Changes nothing about the contents of either.

add_host(host: sys_host.Host) → None#
static assert_is_file(path: str) → None#
abstractmethod available_formats() → list[str]#
async boot_artifacts(inst: inst_base.Instantiation, kinds: list[BootArtifact]) → dict[BootArtifact, str]#

Boot files for this image, from the cache when a previous run put them there. That is what makes them survive a cache hit, where no build runs and a backend that collects them while building never gets the chance.

Each kind is looked for in that order: boot_artifact_paths, then the cache, then _produce_boot_artifacts.

static file_identity(path: str) → str#

Cheap stand-in for hashing a whole disk image, which is far too big.

Catches the file being replaced or rewritten, which is what matters for an image someone drops into place.

find_format(host: sim_host.HostSim) → str#
id() → int#
path(inst: inst_base.Instantiation, format: str) → str#
async prepare(inst: inst_base.Instantiation, host: sys_host.Host) → None#
toJSON() → dict#
boot_artifact_paths: dict[BootArtifact, str]#

Boot artifacts supplied as files on the runner, by kind, overriding whatever the image itself would give. Not stored in the cache.

virt_ls_exec#

Lists /boot in the built image to find the kernel version its boot artifacts are named after.

virt_copy_out_exec#

Copies the boot artifacts out of the built image.

classmethod fromJSON(system: sys_base.System, json_obj: dict) → tpe.Self#