1
0
Fork 0

generate: container: Implement Accounting, Cgroup and Cpu

Implment the 'Accounting', 'Cgroup', and 'Cpu' classes to control the
contaienr cgroup and cpu usage.

Signed-off-by: Enno Tensing <tenno@suij.in>
This commit is contained in:
Enno Tensing 2025-07-30 10:24:51 +02:00
parent 57b983e876
commit 2a7bb0115c
Signed by: tenno
GPG key ID: 95265603BD36E66C

View file

@ -29,7 +29,7 @@ class ConfigError(Exception):
return f"Configuration error: {self.message}" return f"Configuration error: {self.message}"
def maybe(json: dict, key: str) -> str | dict | list | bool | None: def maybe(json: dict, key: str) -> ConfigValue:
"""Maybe get a value.""" """Maybe get a value."""
try: try:
return json[key] return json[key]
@ -37,6 +37,14 @@ def maybe(json: dict, key: str) -> str | dict | list | bool | None:
return None return None
def maybe_or(json: dict, key: str, _or: ConfigValue) -> ConfigValue:
"""Maybe get a value, but return _or if it is None."""
val = maybe(json, key)
if val is None or not isinstance(val, _or):
return _or
return val
def trim(s: str) -> str: def trim(s: str) -> str:
"""Remove sequential whitespace.""" """Remove sequential whitespace."""
s = s.replace("\t ", "\t") s = s.replace("\t ", "\t")
@ -46,6 +54,149 @@ def trim(s: str) -> str:
return s return s
@dataclass
class Cgroup:
"""cgroup Config."""
config: list
parent: str
namespace: str
how: str
@classmethod
def from_json(cls, val: ConfigValue, logger: Log) -> Self:
"""Create from JSON."""
if not isinstance(val, dict):
logger.log_warning("cgroup Config is invalid!")
return cls([], "", "", "")
config = maybe_or(val, "config", [])
parent = maybe_or(val, "parent", "")
namespace = maybe_or(val, "namespace", "")
how = maybe_or(val, "how", "")
if not isinstance(config, list):
logger.log_warning("Config key in cgroup Config is invalid!")
config = []
if not isinstance(parent, str):
logger.log_warning("Parent key in cgroup Config is invalid!")
parent = ""
if not isinstance(namespace, str):
logger.log_warning("Namespace key in cgroup Config is invalid!")
namespace = ""
if not isinstance(how, str):
logger.log_warning("How key in cgroup Config is invalid!")
how = ""
if how == "split" and parent != "":
logger.log_warning(
"Split cgroups can not be combined with a cgroup parent!"
)
parent = ""
return cls(config, parent, namespace, how)
def command(self) -> str:
"""Option for podman container create."""
cmd = ""
seperator = " \\\n"
cmd += seperator.join(
[f"\t--cgroup-conf={option}" for option in self.config]
)
if self.parent != "":
cmd += f"\t--cgroup-parent={self.parent}{seperator}"
if self.namespace != "":
cmd += f"\t--cgroupns={self.namespace}{seperator}"
if self.how != "":
cmd += f"\t--cgroups={self.how}{seperator}"
return cmd
@dataclass
class Cpu:
"""CPU Accounting."""
period: str
quota: str
shares: str
number: str
cpuset_cpus: str
cpuset_mems: str
@classmethod
def from_json(cls, val: ConfigValue, logger: Log) -> Self:
"""Create from JSON."""
if not isinstance(val, dict):
logger.log_warning("cpu Config is invalid!")
return cls("", "", "", "", "", "")
period = maybe_or(val, "period", "")
quota = maybe_or(val, "quota", "")
shares = maybe_or(val, "shares", "")
number = maybe_or(val, "number", "")
cpuset = maybe_or(val, "cpuset", {})
cpuset_cpus = ""
cpuset_mems = ""
if len(cpuset) != 0:
cpuset_cpus += maybe_or(cpuset, "cpus", "")
cpuset_mems += maybe_or(cpuset, "mems", "")
return cls(
period,
quota,
shares,
number,
cpuset_cpus,
cpuset_mems,
)
def command(self) -> str:
"""Option for podman container create."""
cmd = ""
seperator = " \\\n"
if self.period != "":
cmd += f"\t--cpu-period={self.period}{seperator}"
if self.quota != "":
cmd += f"\t--cpu-quota={self.quota}{seperator}"
if self.shares != "":
cmd += f"\t--cpu-shares={self.shares}{seperator}"
if self.number != "":
cmd += f"\t--cpus={self.number}{seperator}"
if self.cpuset_cpus != "":
cmd += f"\t--cpuset-cpus={self.cpuset_cpus}{seperator}"
if self.cpuset_mems != "":
cmd += f"\t--cpuset-mems={self.cpuset_mems}{seperator}"
return cmd
@dataclass
class Accounting:
"""Resource Accounting."""
cgroup: Cgroup
cpu: Cpu
@classmethod
def from_json(cls, data: ConfigValue, logger: Log) -> Self:
"""Create from JSON."""
cgroup_data = maybe(data, "cgroup")
cpu_data = maybe(data, "cpu")
cgroup = Cgroup.from_json(cgroup_data, logger)
cpu = Cpu.from_json(cpu_data, logger)
return cls(cgroup, cpu)
def command(self) -> str:
"""Options for podman container create."""
cgroup = self.cgroup.command()
cpu = self.cpu.command()
return cgroup + cpu
@dataclass @dataclass
class Volume: class Volume:
"""Container Volume.""" """Container Volume."""
@ -136,9 +287,9 @@ class Secret:
def command(self) -> str: def command(self) -> str:
"""Option for podman container create.""" """Option for podman container create."""
cmd = ( cmd = (
f"--secret {self.name},:" f"--secret {self.name},:"
f"type={self.secret_type}," f"type={self.secret_type},"
f"target={self.target}" f"target={self.target}"
) )
# Not a password, ruff... # Not a password, ruff...
if self.secret_type == "mount" and self.options != "": # noqa: S105 if self.secret_type == "mount" and self.options != "": # noqa: S105