tessl/pypi-cgroupspy

Python library for managing Linux control groups (cgroups) with pythonic interface to filesystem operations

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Resource Controllers

Name: tessl/pypi-cgroupspy
Author: tessl

Controller classes providing typed access to cgroup resource management files. Controllers automatically handle data conversion between Python objects and cgroup-compatible strings, enabling pythonic management of system resources.

Capabilities

Base Controller

Base controller class providing common functionality for all cgroup property management with file interface abstractions.

class Controller:
    """Base controller for cgroup property access and file management."""
    
    # Common properties available to all controllers
    tasks: List[int]  # Task IDs in cgroup
    procs: List[int]  # Process IDs in cgroup
    notify_on_release: bool  # Notify on release flag
    clone_children: bool  # Clone children flag
    
    def __init__(self, node):
        """
        Initialize controller for a specific node.
        
        Parameters:
        - node: Node, the cgroup node this controller manages
        """
    
    def filepath(self, filename: str) -> bytes:
        """
        Get full path to a cgroup control file.
        
        Parameters:
        - filename: str, name of the control file
        
        Returns:
        Full filesystem path to the file
        """
    
    def list_interfaces(self) -> list:
        """
        List all available file interfaces for this controller.
        
        Returns:
        List of interface names available for this controller
        """
    
    def get_interface(self, key):
        """
        Get file interface object by key name.
        
        Parameters:
        - key: str, interface key name
        
        Returns:
        BaseFileInterface object for the specified interface
        """
    
    def get_property(self, filename: str) -> str:
        """
        Read raw string value from a cgroup control file.
        
        Parameters:
        - filename: str, control file name
        
        Returns:
        Raw string content of the file
        
        Raises:
        OSError if file doesn't exist or cannot be read
        """
    
    def get_content(self, key):
        """
        Get processed content using file interface by key.
        
        Parameters:
        - key: str, interface key name
        
        Returns:
        Processed content based on interface type (int, dict, list, etc.)
        """
    
    def set_property(self, filename: str, value) -> int:
        """
        Write value to a cgroup control file.
        
        Parameters:
        - filename: str, control file name
        - value: any, value to write (converted to string)
        
        Returns:
        Number of bytes written
        
        Raises:
        OSError if file cannot be written
        """

CPU Controller

CPU scheduling and bandwidth management with CFS (Completely Fair Scheduler) and RT (Real-Time) parameters.

class CpuController(Controller):
    """CPU cgroup controller for scheduling and bandwidth management."""
    
    # CFS (Completely Fair Scheduler) parameters
    cfs_period_us: int  # CFS period in microseconds
    cfs_quota_us: int   # CFS quota in microseconds (-1 for unlimited)
    
    # RT (Real-Time) parameters
    rt_period_us: int   # RT period in microseconds
    rt_runtime_us: int  # RT runtime in microseconds
    
    # CPU shares for proportional scheduling
    shares: int         # CPU shares (default: 1024)
    
    # Statistics
    stat: dict          # CPU usage statistics

Usage example:

from cgroupspy.trees import Tree

# Get CPU controller
t = Tree()
cpu_node = t.get_node_by_path('/cpu/test/')

# Set CPU quota (50% of one CPU core)
cpu_node.controller.cfs_period_us = 100000  # 100ms
cpu_node.controller.cfs_quota_us = 50000    # 50ms

# Set CPU shares (relative weight)
cpu_node.controller.shares = 512  # Half the default weight

# Check statistics
stats = cpu_node.controller.stat
print(stats['nr_periods'])    # Number of periods
print(stats['nr_throttled'])  # Number of throttled periods

CPU Accounting Controller

CPU usage tracking and per-CPU statistics for monitoring and accounting purposes.

class CpuAcctController(Controller):
    """CPU accounting controller for usage tracking."""
    
    # Usage statistics
    usage: int                # Total CPU usage in nanoseconds
    usage_percpu: List[int]   # Per-CPU usage in nanoseconds
    
    # Detailed statistics
    stat: dict               # CPU accounting statistics (user/system time)

Usage example:

from cgroupspy.trees import Tree

# Get CPU accounting controller
t = Tree()
cpuacct_node = t.get_node_by_path('/cpuacct/test/')

# Read usage statistics
total_usage = cpuacct_node.controller.usage  # Total nanoseconds
per_cpu = cpuacct_node.controller.usage_percpu  # Per-CPU usage

# Get detailed statistics
stat = cpuacct_node.controller.stat
print(f"User time: {stat['user']}")
print(f"System time: {stat['system']}")

CPU Set Controller

CPU and memory node assignment for NUMA-aware process placement and CPU pinning.

class CpuSetController(Controller):
    """CPU set controller for CPU and memory node assignment."""
    
    # Core assignments
    cpus: set              # Set of allowed CPU cores
    mems: set              # Set of allowed memory nodes
    
    # Inheritance and spreading flags
    cpu_exclusive: bool           # Exclusive CPU access
    mem_exclusive: bool           # Exclusive memory access
    mem_hardwall: bool           # Hard memory wall
    memory_migrate: bool         # Migrate memory on CPU change
    memory_pressure_enabled: bool # Enable memory pressure notifications
    memory_spread_page: bool     # Spread page cache
    memory_spread_slab: bool     # Spread slab cache
    sched_load_balance: bool     # Enable scheduler load balancing
    sched_relax_domain_level: int # Scheduler domain relaxation level

Usage example:

from cgroupspy.trees import Tree

# Get CPU set controller
t = Tree()
cpuset_node = t.get_node_by_path('/cpuset/test/')

# Pin to specific CPUs
cpuset_node.controller.cpus = {0, 1, 2, 3}  # CPUs 0-3
cpuset_node.controller.mems = {0}           # Memory node 0

# Enable exclusive access
cpuset_node.controller.cpu_exclusive = True
cpuset_node.controller.mem_exclusive = True

# Configure memory policies
cpuset_node.controller.memory_migrate = True
cpuset_node.controller.memory_spread_page = True

Memory Controller

Memory usage limits, statistics, and OOM (Out-of-Memory) control for resource management.

class MemoryController(Controller):
    """Memory controller for usage limits and statistics."""
    
    # Memory limits and usage
    limit_in_bytes: int          # Memory limit in bytes (-1 for unlimited)
    usage_in_bytes: int          # Current memory usage in bytes
    max_usage_in_bytes: int      # Maximum memory usage reached
    
    # Kernel memory (if enabled)
    kmem_limit_in_bytes: int     # Kernel memory limit
    kmem_usage_in_bytes: int     # Kernel memory usage
    
    # Swap control
    memsw_limit_in_bytes: int    # Memory + swap limit
    memsw_usage_in_bytes: int    # Memory + swap usage
    memsw_max_usage_in_bytes: int # Maximum memory + swap usage
    
    # OOM control
    oom_control: dict            # OOM killer control settings
    
    # Memory pressure notifications
    pressure_level: int          # Memory pressure level
    
    # Statistics
    stat: dict                   # Detailed memory statistics
    
    # Control flags
    swappiness: int             # Swappiness value (0-100)
    move_charge_at_immigrate: int # Charge movement on task migration
    use_hierarchy: bool         # Use hierarchical accounting

Usage example:

from cgroupspy.trees import Tree

# Get memory controller
t = Tree()
mem_node = t.get_node_by_path('/memory/test/')

# Set memory limit (1GB)
mem_node.controller.limit_in_bytes = 1024 * 1024 * 1024

# Set swap limit (2GB total)
mem_node.controller.memsw_limit_in_bytes = 2 * 1024 * 1024 * 1024

# Configure swappiness
mem_node.controller.swappiness = 10  # Low swappiness

# Check current usage
current = mem_node.controller.usage_in_bytes
limit = mem_node.controller.limit_in_bytes
print(f"Memory usage: {current / (1024**3):.2f} GB / {limit / (1024**3):.2f} GB")

# Check OOM status
oom = mem_node.controller.oom_control
print(f"OOM disabled: {oom['oom_kill_disable']}")

Device Controller

Device access control with allow/deny rules for character and block devices.

class DevicesController(Controller):
    """Device access controller for device permissions."""
    
    # Device access control
    allow: str    # Allow device access (write-only)
    deny: str     # Deny device access (write-only)  
    list: str     # List current device permissions (read-only)

Usage example:

from cgroupspy.trees import Tree
from cgroupspy.contenttypes import DeviceAccess

# Get devices controller
t = Tree()
devices_node = t.get_node_by_path('/devices/test/')

# Allow access to specific device
# Format: "type major:minor permissions"
devices_node.controller.allow = "c 1:5 rwm"  # /dev/console

# Using DeviceAccess helper
access = DeviceAccess(
    dev_type=DeviceAccess.TYPE_CHAR,
    major=1, minor=5,
    access=DeviceAccess.ACCESS_READ | DeviceAccess.ACCESS_WRITE
)
devices_node.controller.allow = str(access)

# Deny access to all devices
devices_node.controller.deny = "all"  # Deny all

# List current permissions
permissions = devices_node.controller.list
print(permissions)

Block I/O Controller

Block device I/O throttling, bandwidth control, and statistics for storage resource management.

class BlkIOController(Controller):
    """Block I/O controller for bandwidth control and throttling."""
    
    # Weight-based bandwidth control
    weight: int                    # I/O weight (100-1000)
    weight_device: str            # Per-device I/O weights
    
    # Throttling controls
    throttle_read_bps_device: str   # Read bandwidth throttling
    throttle_write_bps_device: str  # Write bandwidth throttling
    throttle_read_iops_device: str  # Read IOPS throttling
    throttle_write_iops_device: str # Write IOPS throttling
    
    # Statistics
    time: dict                     # I/O time statistics
    sectors: dict                  # I/O sector statistics
    io_service_bytes: dict         # I/O service bytes
    io_serviced: dict             # I/O operations serviced
    io_service_time: dict         # I/O service time
    io_wait_time: dict            # I/O wait time
    io_merged: dict               # I/O merged operations
    io_queued: dict               # I/O queued operations
    
    # Reset controls
    reset_stats: int              # Reset statistics (write-only)

Usage example:

from cgroupspy.trees import Tree

# Get Block I/O controller
t = Tree()
blkio_node = t.get_node_by_path('/blkio/test/')

# Set I/O weight
blkio_node.controller.weight = 500  # Medium priority

# Throttle read bandwidth (10MB/s on device 8:0)
blkio_node.controller.throttle_read_bps_device = "8:0 10485760"

# Throttle write IOPS (1000 IOPS on device 8:0)
blkio_node.controller.throttle_write_iops_device = "8:0 1000"

# Check I/O statistics
stats = blkio_node.controller.io_service_bytes
for device, bytes_stats in stats.items():
    print(f"Device {device}: Read {bytes_stats['Read']} bytes")

Network Classifier Controller

Network packet classification tagging for Quality of Service (QoS) and traffic shaping integration.

class NetClsController(Controller):
    """Network classifier controller for packet tagging."""
    
    classid: int  # Network class ID for packet tagging

Usage example:

from cgroupspy.trees import Tree

# Get network classifier controller
t = Tree()
netcls_node = t.get_node_by_path('/net_cls/test/')

# Set class ID for packet tagging
netcls_node.controller.classid = 0x10001  # Class ID for tc/iptables

Network Priority Controller

Network interface priority mapping for controlling network bandwidth allocation per interface.

class NetPrioController(Controller):
    """Network priority controller for interface priority mapping."""
    
    prioidx: int        # Priority index
    ifpriomap: dict     # Interface priority mapping

Usage example:

from cgroupspy.trees import Tree

# Get network priority controller
t = Tree()
netprio_node = t.get_node_by_path('/net_prio/test/')

# Set priority index
netprio_node.controller.prioidx = 1

# Set interface priorities
netprio_node.controller.ifpriomap = {
    "eth0": 5,  # High priority
    "eth1": 1   # Low priority
}

Types

# File interface base classes
class BaseFileInterface:
    """Base interface for cgroup file access with type conversion."""
    readonly: bool
    writeonly: bool
    
    def __init__(self, filename: str, readonly=None, writeonly=None): ...
    def sanitize_get(self, value: str): ...
    def sanitize_set(self, value): ...

class FlagFile(BaseFileInterface):
    """Boolean flag file interface (0/1 to False/True)."""

class IntegerFile(BaseFileInterface):
    """Integer file interface with automatic conversion."""

class DictFile(BaseFileInterface):
    """Dictionary file interface for key-value files."""

class ListFile(BaseFileInterface):
    """List file interface for space-separated values."""

class IntegerListFile(ListFile):
    """Integer list file interface."""

class CommaDashSetFile(BaseFileInterface):
    """Set file interface for comma-dash ranges (e.g., '0-2,4,7-9')."""

class MultiLineIntegerFile(BaseFileInterface):
    """Multi-line integer file interface (one integer per line)."""

class TypedFile(BaseFileInterface):
    """Content type file interface for complex data structures."""

# Content type classes for specialized data
class DeviceAccess:
    """Device access permissions with type and access flags."""
    TYPE_ALL: str = "all"
    TYPE_CHAR: str = "c"
    TYPE_BLOCK: str = "b"
    ACCESS_READ: int = 1
    ACCESS_WRITE: int = 2
    ACCESS_MKNOD: int = 4
    
    def __init__(self, dev_type=None, major=None, minor=None, access=None): ...
    @property
    def can_read(self) -> bool: ...
    @property
    def can_write(self) -> bool: ...
    @property
    def can_mknod(self) -> bool: ...
    @property
    def access_string(self) -> str: ...

class DeviceThrottle:
    """Device throttling limits for bandwidth/IOPS control."""
    def __init__(self, limit: int, major=None, minor=None): ...

Install with Tessl CLI