SWAMPy

✎ CARVE FLESH

EDITING: commands.py

# 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 (
    PAGE_SIZE, DomainEntry, LvdConfig,
    config_path, get_username, resolve_docroot,
)
from .exceptions import ErrorCode, 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'

# Exit status of the SUID helpers -> error code, so a failure is classified
# without parsing the helper's message. Keep in sync with the EXIT_* values
# in src/lvd-helpers/lvd-{limits,registry}-helper.c.
HELPER_EXIT_CODES = {
    2: ErrorCode.INVALID_VALUE,
    3: ErrorCode.NOT_ALLOWED,
    4: ErrorCode.NOT_ISOLATED,
    5: ErrorCode.BACKEND_UNAVAILABLE,
    6: ErrorCode.KERNEL_ERROR,
}

def _helper_error(helper, result):
    """LvdError for a helper that exited non-zero, coded by its exit status."""
    name = os.path.basename(helper)
    return LvdError(
        f"{name} failed: {result.stderr.strip()}",
        code=HELPER_EXIT_CODES.get(result.returncode, ErrorCode.BACKEND_UNAVAILABLE))

# 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}",
                       code=ErrorCode.BACKEND_UNAVAILABLE) from exc
    return {name for name, _docroot in pairs}

def _load_config(lve_id):
    """Load the account's config, refusing to act on one that could not be read.

A config that could not be read is EMPTY, so without this every command
    would answer as though the account simply had no per-domain limits: `list`
    reports none, `apply` reports a configured domain as never configured, and
    `set` persists that emptiness over the file (CLOS-7494). One wrong answer
    is a wrong diagnosis, the other loses the records -- neither is worth
    giving when the real state is "this file needs attention".
    """
    config = LvdConfig.load(lve_id)
    if config.unreadable:
        raise LvdError(
            f"cannot read {config_path(lve_id)}; it may hold limits that this "
            f"command would otherwise ignore or discard. See "
            f"~/.lve/isolatectl.log for what is wrong with it, then repair or "
            f"remove the file", code=ErrorCode.CONFIG_UNREADABLE)
    return config

def _docroot_for(domain):
    """Resolve domain -> docroot."""
    docroot = resolve_docroot(domain)
    if not docroot:
        raise LvdError(f"cannot resolve document root for domain '{domain}'",
                       code=ErrorCode.BACKEND_UNAVAILABLE)
    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}",
                       code=ErrorCode.BACKEND_UNAVAILABLE) from e

_dbg(f"  rc={result.returncode} stdout={result.stdout.strip()!r}"
         f" stderr={result.stderr.strip()!r}")

if result.returncode != 0:
        raise _helper_error(REGISTRY_HELPER, result)

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 // PAGE_SIZE if pmem_bytes else 0
    vmem_bytes = limits.get('vmem', 0)
    vmem_pages = vmem_bytes // PAGE_SIZE 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}",
                       code=ErrorCode.BACKEND_UNAVAILABLE) 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:
        raise _helper_error(LIMITS_HELPER, result)

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}",
                       code=ErrorCode.DOMAIN_NOT_OWNED)

# 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",
            code=ErrorCode.NOT_ISOLATED,
        )

config = _load_config(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()

# Kernel first, config second: a value the kernel refuses must not be
    # stored, or domains.json would claim a limit that is not in force and
    # the next apply would retry it (CLOS-7520). If the save then fails, put
    # the previous limits back so the kernel does not run ahead of the file.
    _call_limits_helper(lve_id, domain_id, new_limits)
    try:
        config.save()
    except (LvdError, OSError):
        if old_limits:
            try:
                _call_limits_helper(lve_id, domain_id, old_limits)
            except LvdError as exc:
                log.warning("cannot restore previous limits of '%s': %s", domain, exc)
        else:
            # No limits of its own before: the domain LVE ran on the account's
            # defaults, which the helper cannot write back -- it would zero
            # every limit instead. Keep the limits just applied; the next
            # `lvectl apply` resets the LVE to the defaults.
            log.warning("'%s' keeps the limits just applied until the next "
                        "lvectl apply: its configuration could not be saved", domain)
        raise

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)

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 = _load_config(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}",
                       code=ErrorCode.BACKEND_UNAVAILABLE) 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',
                       code=ErrorCode.BACKEND_UNAVAILABLE)

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}",
            code=ErrorCode.BACKEND_UNAVAILABLE,
        ) 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}",
                       code=ErrorCode.DOMAIN_NOT_OWNED)

config = _load_config(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/<uid> disagree, so say so
        # rather than reporting a bare null.
        #
        # WARNING, deliberately not ERROR. Two reasons, either sufficient:
        #
        # 1. It could not reach Sentry anyway. The only caller is cmd_list, and
        #    the only way in is isolatectl, which refuses to run as root,
        #    initialises no Sentry client, and sets websiteisolation.propagate
        #    to False. An ERROR here would be a Sentry escalation that never
        #    escalates to anything.
        # 2. It is the same condition _build_domain_map deliberately DROPPED to
        #    DEBUG: a domain with no registry id is one the administrator has
        #    not run enable-domain-limits for, which is the steady state of
        #    every partial rollout. Reporting it here while ignoring it there
        #    would contradict the rest of this change.
        #
        # It stays a WARNING because it is still worth having in the tenant's
        # own isolatectl.log when a listing shows a 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",
                       code=ErrorCode.INVALID_VALUE)

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",
            code=ErrorCode.NOT_ISOLATED,
        )

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)

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-11-02 22:52
📄 __init__.py	PY	1.5 KB	2026-10-16 11:34	EDIT
📄 commands.py	PY	18.4 KB	2026-10-16 11:34	EDIT
📄 config.py	PY	25.9 KB	2026-10-16 11:34	EDIT
📄 exceptions.py	PY	1.4 KB	2026-10-16 11:34	EDIT
📄 id_registry.py	PY	5.2 KB	2026-10-16 11:34	EDIT
📄 limits.py	PY	1.8 KB	2026-10-16 11:34	EDIT