✏️ Editing: net_acct.py Read Only

# coding=utf-8
#
# 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
"""
Telemetry collector for LVE Traffic Accounting (CLOS-4341).

Parses /proc/lve/list and produces a small set of metrics that let us answer:
- does this kernel expose net accounting at all (NETO/NETI columns present)?
- how many user LVEs exist and how many actually accumulated traffic?
- what's the cumulative NETO/NETI volume across user LVEs on this host?

/proc/lve/list contains two non-user LVE rows we exclude from all aggregates:
- *default* (user_id = UINT_MAX) — kernel catch-all for processes not
  attached to any specific LVE (sshd, systemd, dnf, ...). Has non-zero
  traffic on essentially every running CL host.
- *root* (user_id = 0) — LVE container for root (uid 0). Not a hosting user.

After exclusion the metrics describe user hosting activity, and lves_total
matches the count seen by `lvectl list` and the panel's user list.
"""

import os
from typing import Dict, Iterable, Optional

PROC_LVE_LIST = "/proc/lve/list"

# user-id slots that don't represent hosting customers and are excluded from
# all per-LVE aggregates: default catch-all bucket and root.
NON_USER_LVE_IDS = frozenset({0, 0xFFFFFFFF})  # 0 = root, UINT_MAX = default

METRIC_NAMES = (
    "net_acct_kernel_supported",
    "net_acct_lves_total",
    "net_acct_lves_with_traffic",
    "net_acct_total_neto_bytes",
    "net_acct_total_neti_bytes",
)

def _user_id_excluded(lve_id_field: str, excluded: Iterable[int]) -> bool:
    """True if the LVE row's user_id is in the excluded set.

LVE ids in /proc/lve/list use the form "<lvp_id>,<user_id>" (e.g. "0,1002")
    or just "<user_id>" on older kernels. Malformed rows are excluded.
    """
    user_id = lve_id_field.rsplit(",", 1)[-1]
    try:
        return int(user_id) in excluded
    except ValueError:
        return True

def _empty_result(supported: int = 0) -> Dict[str, int]:
    return {
        "net_acct_kernel_supported": supported,
        "net_acct_lves_total": 0,
        "net_acct_lves_with_traffic": 0,
        "net_acct_total_neto_bytes": 0,
        "net_acct_total_neti_bytes": 0,
    }

def parse_proc_lve_list(
    content: str,
    skip_user_ids: Optional[Iterable[int]] = None,
) -> Dict[str, int]:
    """Parse /proc/lve/list text and return the 5 net_acct metrics.

Format of /proc/lve/list:
        <version>:<TAB>LVE<TAB>...<TAB>NETO<TAB>NETI    # header
        <lve_id><TAB>...<TAB><neto><TAB><neti>          # one row per LVE

The leading "<version>:" prefix is optional/version-dependent; we tolerate
    its presence and absence. NETO/NETI are cumulative byte counters; their
    lowercase counterparts lNETO/lNETI are limit columns and ignored here.

skip_user_ids — additional user-id slots to exclude beyond the standard
    root/default. Used by cloudlinux-summary to skip its own transient
    self-LVE (created via _run_self_in_lve) which would otherwise inflate
    counts by 1 on every collection run.

On any structural problem (no header, missing NETO/NETI columns, malformed
    rows) we treat the kernel as not supporting net accounting and return a
    zeroed result.
    """
    if not content:
        return _empty_result()

lines = content.splitlines()
    if not lines:
        return _empty_result()

header = lines[0]
    # Drop optional "<version>:" prefix that lvectl-format headers carry.
    if ":" in header:
        header = header.split(":", 1)[1]

columns = header.split("\t")
    try:
        neto_idx = columns.index("NETO")
        neti_idx = columns.index("NETI")
    except ValueError:
        return _empty_result()

excluded_ids = set(NON_USER_LVE_IDS)
    if skip_user_ids:
        excluded_ids.update(skip_user_ids)

total = 0
    with_traffic = 0
    sum_neto = 0
    sum_neti = 0

for raw in lines[1:]:
        if not raw.strip():
            continue
        fields = raw.split("\t")
        if len(fields) <= max(neto_idx, neti_idx):
            continue
        if _user_id_excluded(fields[0], excluded_ids):
            continue
        try:
            neto = int(fields[neto_idx])
            neti = int(fields[neti_idx])
        except ValueError:
            continue

total += 1
        if neto > 0 or neti > 0:
            with_traffic += 1
        sum_neto += neto
        sum_neti += neti

return {
        "net_acct_kernel_supported": 1,
        "net_acct_lves_total": total,
        "net_acct_lves_with_traffic": with_traffic,
        "net_acct_total_neto_bytes": sum_neto,
        "net_acct_total_neti_bytes": sum_neti,
    }

def collect_net_acct_metrics(
    path: str = PROC_LVE_LIST,
    skip_user_ids: Optional[Iterable[int]] = None,
) -> Dict[str, int]:
    """Read /proc/lve/list and return parsed metrics.

Missing file (older kernels, non-CL kernels, Ubuntu without lve) yields
    kernel_supported=0 and zeros for the rest.
    """
    if not os.path.exists(path):
        return _empty_result()
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            content = fh.read()
    except OSError:
        return _empty_result()
    return parse_proc_lve_list(content, skip_user_ids=skip_user_ids)

🔐 Sid Gifari File Manager Pro