/opt/cloudlinux/venv/lib/python3.11/site-packages/websiteisolation
Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/websiteisolation/commands.py (14589B)
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
"""Public command API for isolatectl limits — per-domain (LVD) resource limit management."""
import json
import logging
import os
import subprocess
import sys
import syslog
from clcommon.cpapi import userdomains
from .config import (
DomainEntry, LvdConfig,
get_username, resolve_docroot,
)
from .exceptions import LvdError
log = logging.getLogger(__name__)
REGISTRY_HELPER = '/usr/share/lve-utils/lvd-registry-helper'
LIMITS_HELPER = '/usr/share/lve-utils/lvd-limits-helper'
STATS_HELPER = '/usr/share/lve-utils/lvd-stats-helper'
# lveinfo's per-domain columns, grouped the way a reader consumes them. Each
# tuple is (our key, lveinfo's column) and the four groups share one key set, so
# a caller renders "14.7 of 100, hit the cap 31 times" without special-casing a
# metric.
#
# Deliberately flat rather than cloudlinux-statistics' {"cpu": {"lve": ...}}:
# that nesting exists to carry a MySQL Governor companion, and governor
# accounting is per account, so per domain it would be a permanently one-keyed
# object promising a sibling that cannot exist.
_STAT_GROUPS = (
('usage', 'a'), # aCPU, aEP, ... average over the window
('peak', 'm'), # mCPU, mEP, ... highest sample in it
('limits', 'l'), # lCPU, lEP, ... the cap in force
)
# (our key, lveinfo's suffix-less name). lveinfo spells faults inconsistently —
# EPf and CPUf but VMemF and PMemF — so the mapping is spelled out rather than
# derived.
_STAT_METRICS = (
('cpu', 'CPU'),
('ep', 'EP'),
('vmem', 'VMem'),
('pmem', 'PMem'),
('nproc', 'Nproc'),
('io', 'IO'),
('iops', 'IOPS'),
)
_FAULT_COLUMNS = {
'cpu': 'CPUf',
'ep': 'EPf',
'vmem': 'VMemF',
'pmem': 'PMemF',
'nproc': 'NprocF',
'io': 'IOf',
'iops': 'IOPSf',
}
_DEBUG = int(os.environ.get('PYLVE_DEBUG', 0))
def _ok(**kwargs):
return {'result': 'success', **kwargs}
def _user_domains(lve_id):
"""Return set of domain names that belong to the user (via panel API)."""
username = get_username(lve_id)
try:
pairs = userdomains(username) or []
except Exception as exc:
raise LvdError(f"failed to query domains for user '{username}': {exc}") from exc
return {name for name, _docroot in pairs}
def _docroot_for(domain):
"""Resolve domain -> docroot."""
docroot = resolve_docroot(domain)
if not docroot:
raise LvdError(f"cannot resolve document root for domain '{domain}'")
return docroot
def _helper_env():
"""Build environment for SUID helper subprocesses."""
env = os.environ.copy()
if _DEBUG:
env['LIBLVE_DEBUG_ENABLED'] = '1'
return env
def _dbg(msg):
if _DEBUG:
print(f"DEBUG [lvdctl]: {msg}", file=sys.stderr)
def _get_domain_lve_id(uid, docroot):
"""Call lvd-registry-helper get and return domain_id, or None if not found."""
argv = [REGISTRY_HELPER, 'get', str(uid), docroot]
_dbg(f"call {REGISTRY_HELPER} get uid={uid} docroot={docroot}")
try:
result = subprocess.run(
argv, capture_output=True, text=True, check=False,
env=_helper_env(),
)
except OSError as e:
raise LvdError(f"failed to run {REGISTRY_HELPER}: {e}") from e
_dbg(f" rc={result.returncode} stdout={result.stdout.strip()!r}"
f" stderr={result.stderr.strip()!r}")
if result.returncode != 0:
stderr = result.stderr.strip()
raise LvdError(f"lvd-registry-helper failed: {stderr}")
out = result.stdout.strip()
if not out:
return None
try:
return int(out)
except ValueError as exc:
raise LvdError(f"lvd-registry-helper returned invalid output: {out!r}") from exc
def _call_limits_helper(uid, domain_id, limits):
"""Call lvd-limits-helper to apply limits to kernel.
Unit conversions (user-facing → kernel):
cpu — centipercent, pass as-is
pmem — bytes → 4 KB pages
io — KB/s, pass as-is
nproc, iops, ep — pass as-is
vmem — bytes → 4 KB pages
"""
pmem_bytes = limits.get('pmem', 0)
pmem_pages = pmem_bytes // 4096 if pmem_bytes else 0
vmem_bytes = limits.get('vmem', 0)
vmem_pages = vmem_bytes // 4096 if vmem_bytes else 0
cpu = limits.get('cpu', 0)
io = limits.get('io', 0)
nproc = limits.get('nproc', 0)
iops = limits.get('iops', 0)
ep = limits.get('ep', 0)
argv = [
LIMITS_HELPER,
str(uid), str(domain_id),
str(cpu), str(pmem_pages), str(io), str(nproc), str(iops),
str(ep), str(vmem_pages),
]
_dbg(f"call {LIMITS_HELPER} uid={uid} domain_id={domain_id}"
f" cpu={cpu} pmem={pmem_pages}pages({pmem_bytes}bytes)"
f" io={io} nproc={nproc} iops={iops}"
f" ep={ep} vmem={vmem_pages}pages({vmem_bytes}bytes)")
try:
result = subprocess.run(
argv, capture_output=True, text=True, check=False,
env=_helper_env(),
)
except OSError as e:
raise LvdError(f"failed to run {LIMITS_HELPER}: {e}") from e
_dbg(f" rc={result.returncode} stderr={result.stderr.strip()!r}")
if result.stdout.strip():
_dbg(f" stdout={result.stdout.strip()!r}")
if result.returncode != 0:
stderr = result.stderr.strip()
raise LvdError(f"lvd-limits-helper failed: {stderr}")
def cmd_set(lve_id, domain, limits):
"""Store per-domain limits in config and apply them to kernel."""
owned = _user_domains(lve_id)
if domain not in owned:
raise LvdError(f"domain '{domain}' does not belong to user with lve_id {lve_id}")
# Verify registration before touching the config: if the domain has no
# assigned LVE ID the limits helper will fail anyway, and we must not
# leave a domains.json entry that can never be applied.
docroot = _docroot_for(domain)
domain_id = _get_domain_lve_id(lve_id, docroot)
if domain_id is None:
raise LvdError(
f"domain '{domain}' has no registered domain ID; "
"the server administrator must run "
f"'lvectl enable-domain-limits {domain}' first"
)
config = LvdConfig.load(lve_id)
entry = config.find_domain(name=domain)
if entry is None:
entry = DomainEntry(name=domain)
config.domains.append(entry)
old_limits = entry.limits.to_dict()
entry.limits.update(**limits)
new_limits = entry.limits.to_dict()
config.save()
try:
syslog.syslog(
syslog.LOG_INFO,
f"lvdctl set: lve_id={lve_id} domain={domain} "
f"old_limits={old_limits} new_limits={new_limits}",
)
except OSError as e:
print(f"lvdctl audit-log syslog failed: {e}", file=sys.stderr)
_call_limits_helper(lve_id, domain_id, new_limits)
return _ok(domain=domain, limits=new_limits)
def cmd_list(lve_id=None, domain=None):
"""
List domains and their limits from config.
Only includes domains that actually belong to the user (via panel API).
``lve_id`` of each row is the per-domain LVE ID the domain's processes
enter; ``owner_uid`` is the user LVE the domain lives under. A domain
that is present in the config but has no registered domain ID reports
``lve_id: null``.
"""
config = LvdConfig.load(lve_id)
owned = _user_domains(lve_id)
domains = config.domains
if domain is not None:
domains = [d for d in domains if d.name == domain]
result = []
for d in domains:
if d.name not in owned:
continue
result.append({
'name': d.name,
'lve_id': _domain_lve_id_or_none(lve_id, d.name),
'owner_uid': lve_id,
'limits': d.limits.to_dict(),
})
return _ok(domains=result)
def _call_stats_helper(period):
"""Run lvd-stats-helper and return lveinfo's parsed JSON.
The helper takes no uid: it reads `getuid()` itself, so there is no
parameter here through which another tenant's figures could be requested.
"""
argv = [STATS_HELPER, 'get', period]
_dbg(f"call {STATS_HELPER} get period={period}")
try:
result = subprocess.run(
argv, capture_output=True, text=True, check=False,
env=_helper_env(),
)
except OSError as e:
raise LvdError(f"failed to run {STATS_HELPER}: {e}") from e
_dbg(f" rc={result.returncode} stderr={result.stderr.strip()!r}")
if result.returncode != 0:
# The helper execs lveinfo, so a non-zero status is usually lveinfo's
# own — an invalid period, or no database to read. Surface its message
# rather than inventing one.
stderr = result.stderr.strip() or result.stdout.strip()
raise LvdError(stderr or 'lvd-stats-helper failed with no output')
try:
return json.loads(result.stdout)
except (json.JSONDecodeError, ValueError) as exc:
raise LvdError(
f"lvd-stats-helper returned output that is not JSON: "
f"{result.stdout.strip()[:200]!r}"
) from exc
def _domain_stats_entry(row):
"""Reshape one lveinfo per-domain row into the isolatectl response shape."""
entry = {
'name': row.get('domain'),
'lve_id': row.get('domain_id'),
'owner_uid': row.get('parent_uid'),
}
for group, prefix in _STAT_GROUPS:
entry[group] = {
key: row.get(f'{prefix}{column}')
for key, column in _STAT_METRICS
}
entry['faults'] = {
key: row.get(column) for key, column in _FAULT_COLUMNS.items()
}
return entry
def cmd_stats(lve_id=None, domain=None, period='10m'):
"""Per-domain usage for the calling user's own isolated domains.
Read-only, and scoped three times over: the helper refuses to report on any
uid but its caller's, `lveinfo --id` scopes in SQL, and the names are
intersected with what the panel still attributes to this user — so a
catalogue row for a domain that has since moved cannot surface here.
``domain`` filters the result rather than being passed down. `lveinfo
--domain` resolves a name across the whole host and switches to per-bucket
series output; since `--id` has already narrowed the rows to this user,
filtering here is both simpler and incapable of looking at anyone else.
An empty ``domains`` list is an ordinary answer, not an error: only domains
active during the window are recorded at all, so a quiet site is absent
rather than reported as zero. ``scope`` says so, so a caller can tell "quiet"
from "no data".
"""
payload = _call_stats_helper(period)
rows = payload.get('data') or []
owned = _user_domains(lve_id)
entries = []
for row in rows:
name = row.get('domain')
if name is None or name not in owned:
# A domain the panel no longer attributes to this user. Not an
# error — the catalogue outlives a transfer — but not ours to show.
if name is not None:
log.warning(
"skipping '%s': the panel does not list it for this user",
name,
)
continue
if domain is not None and name != domain:
continue
entries.append(_domain_stats_entry(row))
return _ok(
scope={
'owner_uid': lve_id,
'period': period,
'note': 'only domains active during the window are recorded',
},
domains=entries,
)
def cmd_apply(lve_id, domain):
"""Push one domain's limits from config to kernel."""
owned = _user_domains(lve_id)
if domain not in owned:
raise LvdError(f"domain '{domain}' does not belong to user with lve_id {lve_id}")
config = LvdConfig.load(lve_id)
return _apply_domain(lve_id, domain, config)
# --- Internal helpers ---
def _domain_lve_id_or_none(lve_id, domain):
"""
Resolve the per-domain LVE ID for one domain, or None when unavailable.
Listing is a read-only report over every domain in the config, so a
single unresolvable domain must not abort the whole listing: a docroot
the panel no longer resolves, or a domain the administrator never
registered with 'lvectl enable-domain-limits', degrades to None and is
logged. The write paths (set/apply) keep raising instead — there the
missing ID means the operation cannot be carried out.
"""
try:
docroot = _docroot_for(domain)
domain_id = _get_domain_lve_id(lve_id, docroot)
except LvdError as exc:
log.warning("cannot resolve domain LVE ID for '%s': %s", domain, exc)
return None
if domain_id is None:
# The registry has no ID for a domain the config claims limits for:
# domains.json and /etc/container/lvd_ids/
disagree, so say so
# rather than reporting a bare null.
log.warning(
"domain '%s' has no registered domain LVE ID; "
"'lvectl enable-domain-limits %s' was never run, or the registry "
"was reset while the stored limits survived", domain, domain)
return domain_id
def _apply_domain(lve_id, domain, config):
"""
Push one domain's limits from config to kernel via SUID helpers.
Looks up the domain ID that was assigned by the admin via
``lvectl enable-domain-limits``. Domain ID assignment is a
root-only operation; users can only read existing mappings and
apply limits to them.
"""
entry = config.find_domain(name=domain)
if entry is None:
raise LvdError(f"domain '{domain}' not found in config; use 'set' first")
docroot = _docroot_for(domain)
domain_id = _get_domain_lve_id(lve_id, docroot)
if domain_id is None:
raise LvdError(
f"domain '{domain}' has no registered domain ID; "
"the server administrator must run "
f"'lvectl enable-domain-limits {domain}' first"
)
applied_limits = entry.limits.to_dict()
try:
syslog.syslog(
syslog.LOG_INFO,
f"lvdctl apply: lve_id={lve_id} domain={domain} "
f"limits={applied_limits}",
)
except OSError as e:
print(f"lvdctl audit-log syslog failed: {e}", file=sys.stderr)
_call_limits_helper(lve_id, domain_id, applied_limits)
return _ok(domain=domain, limits=applied_limits)