Viewing file:  user_limits.py (4.51 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
"""
This module provides functions to retrieve any limits and usage for a specified user.
"""
import subprocess
import json
class LveCommandError(Exception):
def __str__(self):
return f"{self.args[0]}"
def get_lve_limits(username):
"""
Retrieves the LVE limits for a specified user.
Args:
username (str): The username to retrieve LVE limits for.
Returns:
When user limits are found
{
"lve_cpu": {
"limit": int
},
"lve_ep": {
"limit": int
},
"lve_pmem": {
"limit": int
},
"lve_iops": {
"limit": int
},
"lve_io": {
"limit": int
},
"lve_nproc": {
"limit": int
},
}
or an error
{
"error": str
}
"""
cmd = ["/usr/sbin/cloudlinux-limits", "get", "--username", username, "--json"]
try:
stdout = _call_lve_command(cmd)
limits = json.loads(stdout)
except LveCommandError as e:
return {"error": str(e)}
except json.JSONDecodeError as e:
return {"error": f"Failed to decode JSON: {e}"}
if 'users' not in limits or not any(user['username'] == username for user in limits['users']):
return {"error": f"LVE Limit for user {username} not found"}
user_limits = next(user['limits'] for user in limits['users'] if user['username'] == username)
return {
"lve_cpu": {
"limit": int(user_limits['cpu']['all'].strip('*')) # %
},
"lve_ep": {
"limit": int(user_limits['ep'].strip('*')) # proc num
},
"lve_pmem": {
"limit": int(user_limits['pmem'].strip('*')) # byte
},
"lve_iops": {
"limit": int(user_limits['iops'].strip('*')) # op per sec
},
"lve_io": {
"limit": int(user_limits['io']['all'].strip('*')) * 1024 # byte
},
"lve_nproc": {
"limit": int(user_limits['nproc'].strip('*')) # proc num
},
}
def get_lve_usage(username):
"""
Retrieves the current LVE usage for a specified user.
Args:
username (str): The username to retrieve LVE usage for.
Returns:
When user usage are found
{
"lve_cpu": {
"usage": int
},
"lve_ep": {
"usage": int
},
"lve_pmem": {
"usage": int
},
"lve_iops": {
"usage": int
},
"lve_io": {
"usage": int
},
"lve_nproc": {
"usage": int
},
}
or empty dict when user sleep
{}
or an error
{
"error": str
}
"""
cmd = ["/usr/sbin/lveps", "-d", "-o", "id:15,ep:7,pno:7,cpu:7,mem:9,io:11,iops:11"]
try:
stdout = _call_lve_command(cmd)
lines = stdout.strip().split('\n')
except LveCommandError as e:
return {"error": str(e)}
usage = {}
for line in lines[1:]:
fields = line.split()
if fields[0] == username:
usage = {
"lve_cpu": {
"usage": int(fields[3].strip('%')) # (SPEED/CPU) %
},
"lve_ep": {
"usage": int(fields[1]) # (EP) proc num
},
"lve_pmem": {
"usage": int(fields[4]) * 1024 * 1024 # (MEM) byte
},
"lve_iops": {
"usage": int(fields[6]) # (IOPS) op per sec
},
"lve_io": {
"usage": int(fields[5]) # (IO) byte
},
"lve_nproc": {
"usage": int(fields[2]) # (PNO) proc num
},
}
break
return usage
def _call_lve_command(cmd):
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
return result.stdout
except subprocess.CalledProcessError as e:
raise LveCommandError(f"Command '{cmd}' failed with error: {e}")
except Exception as e:
raise LveCommandError(f"An unexpected error occurred: {e}")
|