/var/opt/nydus/ops/customer_local_ops
Edit: /var/opt/nydus/ops/customer_local_ops/__init__.py (35184B)
from datetime import datetime, timedelta
from typing import Any, Dict, List, Tuple, Union
from enum import Enum
import importlib
import json
import logging
import shutil
import sys as _sys
import os as _os
from customer_local_ops.util.constants import CANONICAL_TIMESTRING_FORMAT
from customer_local_ops.util.encryptor import Encryptor
from customer_local_ops.exceptions import EncryptionKeyNotFound
from customer_local_ops.util.execute import runCommand
from customer_local_ops.util.helpers import create_file
from customer_local_ops.util.retry import Retry
from .util import constants as _constants
from .util import encryptor as _encryptor
from .util import sizes as _sizes
__path__.append(_os.path.join(_os.path.dirname(__file__), 'util')) # type: ignore
_sys.modules[__name__ + '.constants'] = _constants
_sys.modules[__name__ + '.encryptor'] = _encryptor
_sys.modules[__name__ + '.sizes'] = _sizes
constants = _constants
encryptor = _encryptor
sizes = _sizes
LOG = logging.getLogger(__name__)
class OpType(Enum):
CONTROL_PANEL = "cp_op"
OPERATING_SYSTEM = "os_op"
CONTROL_PANEL_OPERATING_SYSTEM = "cp_os_op"
UNKNOWN = "unknown"
# TODO: Move to central repo
class ResourceType(Enum):
OVH = "ovh"
OPENSTACK = "openstack"
VIRTUOZZO_VM = "virtuozzo_vm"
OPENSTACK_HOSTINGCLOUD = "openstack_hostingcloud"
RESOURCE_TYPES = list(ResourceType)
NydusResult = Union[Retry, Tuple[bool, Dict[str, Any]], Tuple[bool, Dict[str, Any], int]]
class OpResult:
"""
Class whose objects encapsulate the complete result of an operation.
"""
def __init__(self, success: bool = False, result: Dict[str, Any] = None, retry_interval: int = None):
"""
Initialize an OpResult instance
:param success: Indicates whether or not the operation was successful
:param result: A dictionary containing any messages or other data from the operation
:param retry_interval: An optional retry interval, in seconds. If this value is greater that or equal to
zero, the operation will be retried in that many seconds.
"""
self.success = success
self.result = result
self.retry_interval = retry_interval
@property
def errs(self) -> str:
"""Convenience property for accessing the 'errs` key in the result dict"""
return self.result.get('errs', '') if isinstance(self.result, dict) else ''
@property
def outs(self) -> str:
"""Convenience property for accessing the 'outs` key in the result dict"""
return self.result.get('outs', '') if isinstance(self.result, dict) else ''
def as_tuple(self) -> NydusResult:
"""
Construct and return a correctly-sized tuple, based on whether or not a retry_interval value is present, for
passing back to Nydus.
"""
return ((self.success, self.result) if self.retry_interval is None
else (self.success, self.result, self.retry_interval))
class Ops:
DISK_UTILIZATION_PATH = None
RETRYABLE_ERRS = []
PANOPTA_MANIFEST_FILE = '/etc/panopta-agent-manifest'
DISTRO = None # override
OS_VERSION = None # override
QEMU_PACKAGE_NAME = None # override
def __str__(self):
return self.__class__.__name__
op_type = OpType.UNKNOWN
def get_op_type(self):
return self.op_type
def _get_vm_tag(self):
"""Return this VM's tag. Same return format as runCommand."""
raise NotImplementedError
def _encryption_key(self):
"""The encryption key."""
exit_code, outs, errs = self._get_vm_tag()
if exit_code != 0 or not outs:
LOG.error('Could not obtain encryption key')
raise EncryptionKeyNotFound(exit_code, outs, errs)
return outs
def decrypt(self, encrypted):
"""Decrypt an encrypted value."""
try:
key = self._encryption_key()
return Encryptor.decrypt(encrypted, key)
except Exception:
LOG.exception('Decrypt failed')
raise
def encrypt(self, target: str) -> str:
"""
Encrypt a plain-text value.
:param target: The value to be encrypted
:return: An encrypted string
"""
try:
key = self._encryption_key()
return Encryptor.encrypt(target, key)
except Exception:
LOG.exception('Encrypt failed')
raise
def write_file(self, path, content):
LOG.debug('Writing file: %s', path)
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
def get_os_op(self, os_op: str, control_panel_class_name: str = None) -> "Ops":
"""
Get an instance of an operating system Ops class or ControlPanel-OS Ops class
:param os_op: The operating system class name e.g. linux.CentOS7
:param control_panel_class_name: The control panel class name e.g. cpanel.CPanel
:return: An instance of the operating system class e.g. CentOS7 or CP-OS class e.g. CentOS7CPanel
"""
parts = os_op.split(".")
# Last part of os_op should be the family name followed by the os class name, e.g. linux.CentOS7
os_name = parts[-1]
family_name = parts[-2]
prefix = "customer_local_ops"
if control_panel_class_name is None and self.get_op_type() == OpType.CONTROL_PANEL:
# If we're requesting to get the os op class from a control panel class,
# assume we want the operating system/control panel hybrid class
control_panel_class_name = str(self)
if control_panel_class_name is not None:
# Return the operating system/control panel hybrid class
os_name = os_name + control_panel_class_name
cp_class_name = control_panel_class_name.lower()
module_path = "{prefix}.control_panel.{family_name}_{cp_class_name}".format(prefix=prefix,
family_name=family_name,
cp_class_name=cp_class_name)
else:
module_path = "{prefix}.operating_system.{family_name}".format(prefix=prefix, family_name=family_name)
mod = importlib.import_module(module_path)
class_ = getattr(mod, os_name)
os_op = class_()
return os_op
def build_result_dict(self, outs: str, errs: str, op_name: str) -> Dict[str, Any]:
"""
This function takes command output (outs, errs) and builds and returns a result
dict with op metadata and errors if appropriate
:param outs: The stdout from the run command
:param errs: The stderr from the run command
:param op_name: the name of the op calling this function
:return: a result dict
"""
unicode_null = "\u0000"
outs = outs.replace(unicode_null, "")
errs = errs.replace(unicode_null, "")
result = {'outs': outs, 'errs': errs, 'cls_name': str(self), 'op_type': self.get_op_type().value,
'op_name': op_name}
return result
def get_result_data(self, result: Any) -> OpResult:
"""
This function takes a result tuple or result dict and parses it to a result tuple of a consistent format
that can then be used by ops to inspect the result or build another result based on the first one
:param result: a dict or tuple result from an op
:return: A data structure containing the complete result
"""
if isinstance(result, tuple):
result_len = len(result)
if result_len == 1:
return OpResult(True, result[0])
if result_len == 2:
return OpResult(result[0], result[1])
if result_len == 3:
return OpResult(result[0], result[1], result[2])
# Should never get here
raise ValueError("result tuple has too many elements: {}".format(result_len))
return OpResult(True, result)
def build_result_from_other_result(self, result: Any, op_name: str) -> NydusResult:
"""
This function takes a result tuple/dict/Retry/str from a secondary op (usually an os op) and (if not a Retry)
extracts the metadata, output and errors from it and places it inside a result with metadata from the primary
op (usually a control panel op).
In this way it is clear:
a). what the primary op was and
b). that the error/output is coming from the secondary op
:param result: a dict or tuple result from an op
:param op_name: the name of the op calling this function
:return: a result dict, tuple or Retry instance
"""
if isinstance(result, Retry):
return result
data = self.get_result_data(result)
if isinstance(data.result, dict):
op_type = data.result.get('op_type', self.get_op_type().value)
op_name_other = data.result.get('op_name', op_name)
cls_name = data.result.get('cls_name', str(self))
prefix = "{op_type} {cls_name}.{op_name_other}".format(op_type=op_type, cls_name=cls_name,
op_name_other=op_name_other)
outs = "{prefix} outs: {outs}".format(prefix=prefix, outs=data.outs)
errs = data.errs
if errs:
errs = "{prefix} errs: {errs}".format(prefix=prefix, errs=data.errs)
else:
if data.success:
outs = str(data.result)
errs = ''
else:
outs = ''
errs = str(data.result)
data.result = self.build_result_dict(outs, errs, op_name)
return data.as_tuple()
def build_result_from_cmd_output(self, exit_code: int, outs: str, errs: str, op_name: str,
success_message: str = None) -> NydusResult:
"""
This function takes a result from a command output (exit_code, outs, errs) and builds and returns a result
tuple with op metadata and errors if appropriate
:param exit_code: The exit code returned from the run command
:param outs: The stdout from the run command
:param errs: The stderr from the run command
:param op_name: the name of the op calling this function
:param success_message: A message to use in place of 'outs' if the command ran successfully (optional)
:return: a result tuple
"""
success = exit_code == 0
if success and success_message is not None:
outs = success_message
return success, self.build_result_dict(outs, errs, op_name)
def run_command_and_handle_result(self,
command: Union[List[str], str],
op_name: str,
*args: Any,
intermediate_result: Dict[str, Any] = None,
**kw: Any) -> NydusResult:
"""Run command, return result suitable for a Nydus operation.
:param command: command list or string as per runCommand
:param op_name: name of the operation; used as runCommand tag and result op_name
:param args: additional positional arguments for runCommand
:param intermediate_result: Dict containing metadata for retries
:param kw: additional keyword arguments for runCommand
:return: command result as a Nydus operation result
"""
exit_code, outs, errs = runCommand(command, op_name, *args, **kw)
return self._result_handler(exit_code, outs, errs, op_name, intermediate_result=intermediate_result)
# def configure_ips(self, network_node):
# raise NotImplementedError
# def patch_system(self, payload):
# raise NotImplementedError
# def patch_system_specific_update(self, payload):
# raise NotImplementedError
# def list_vm_patches(self, payload):
# raise NotImplementedError
def add_user(self, payload, unused=None):
raise NotImplementedError
# def remove_user(self, payload):
# raise NotImplementedError
# def change_password(self, payload):
# raise NotImplementedError
# def change_hostname(self, payload):
# raise NotImplementedError
# def install_package(self, payload):
# raise NotImplementedError
# def uninstall_package(self, payload):
# raise NotImplementedError
def shutdown_clean(self, unused=None):
# The `unused` param is an artifact of Archon workflows requiring an I/O
# chain for sequencing.
raise NotImplementedError
# def stop_web_server(self, none):
# raise NotImplementedError
# def start_web_server(self, none):
# raise NotImplementedError
# def prepare_web_server_ssl(self, none):
# raise NotImplementedError
# def deploy_cert(self, payload):
# raise NotImplementedError
# def install_cert(self, payload):
# raise NotImplementedError
# def configure_mta(self, payload):
# raise NotImplementedError
def enable_admin(self, username, unused=None):
# The `unused` param is an artifact of Archon workflows requiring an I/O
# chain for sequencing.
raise NotImplementedError
def disable_admin(self, username, unused=None):
# The `unused` param is an artifact of Archon workflows requiring an I/O
# chain for sequencing.
raise NotImplementedError
def disable_all_admins(self, unused=None):
# The `unused` param is an artifact of Archon workflows requiring an I/O
# chain for sequencing.
raise NotImplementedError
def extend_disk(self, payload, unused=None):
# The `unused` param is an artifact of Archon workflows requiring an I/O
# chain for sequencing.
raise NotImplementedError
def _get_cpu_utilization(self) -> Dict[str, float]:
"""Return current CPU utilization.
:returns: dictionary with keys:
- cpuUsed: percentage in range 0-100
"""
raise NotImplementedError
def _get_disk_utilization(self) -> Dict[str, int]:
"""Return current disk utilization.
:returns: dictionary with keys:
- diskTotal: mebibytes
- diskUsed: mebibytes
"""
disk = shutil.disk_usage(self.DISK_UTILIZATION_PATH)
return {'diskTotal': disk.total >> 20, # B > MiB
'diskUsed': disk.used >> 20}
def _get_memory_utilization(self) -> Dict[str, int]:
"""Return current memory utilization.
:returns: dictionary with keys:
- memoryTotal: mebibytes
- memoryUsed: mebibytes
"""
raise NotImplementedError
def get_utilization(self) -> Dict[str, Union[float, int, str]]:
"""Return current system utilization.
:returns: dictionary with keys:
- collected: current time in
:data:`~customer_local_ops.constants.CANONICAL_TIMESTRING_FORMAT`
- memoryTotal: mebibytes
- memoryUsed: mebibytes
- diskTotal: mebibytes
- diskUsed: mebibytes
- cpuUsed: percentage in range 0-100
"""
result = {}
for stats in (self._get_cpu_utilization(),
self._get_disk_utilization(),
self._get_memory_utilization()):
result.update(stats)
result['collected'] = datetime.utcnow().strftime(
CANONICAL_TIMESTRING_FORMAT)
return result
def _create_panopta_manifest_file(self, payload):
""" Panopta agent uses a manifest file for configuration
of the customer id and templates to use. This method creates that manifest file.
"""
manifest_file = self.PANOPTA_MANIFEST_FILE
customer_key = payload['customer_key']
template_ids = payload['template_ids']
cust_key_contents = """[agent]
customer_key = {customer_key}
templates = {template_ids}
enable_countermeasures = false""".format(customer_key=customer_key, template_ids=template_ids)
server_key = payload.get('server_key')
if server_key:
cust_key_contents += "\nserver_key = {server_key}".format(server_key=server_key)
fqdn = payload.get('fqdn')
if fqdn:
cust_key_contents += "\nfqdn = {fqdn}".format(fqdn=fqdn)
server_name = payload.get('serverName')
if server_name:
cust_key_contents += "\nserver_name = {serverName}".format(serverName=server_name)
disable_server_match = payload.get('disable_server_match', False)
if disable_server_match:
cust_key_contents += "\ndisable_server_match = true"
create_file(manifest_file, cust_key_contents)
def install_panopta(self, payload, *args, **kwargs):
raise NotImplementedError
def delete_panopta(self, *args, **kwargs):
raise NotImplementedError
def get_panopta_server_key(self, *args, **kwargs):
raise NotImplementedError
MAX_METRIC_ITEMS = 500
MAX_METRIC_JSON_SIZE = 50000
TOP_PROCESSES_BY_RESOURCE = 5
INFRA_PACKAGE_EXCLUSIONS = frozenset([
'nydus', 'nydus-executor', 'nydus-ex', 'nydus-ex-api',
'panopta-agent', 'fm-agent', 'fortimonitor-agent',
'cloud-init', 'cloudbase-init', 'waagent', 'vboxguest',
'qemu-guest-agent', 'virtio-win', 'virtio-win-gt-x64',
'gpg-pubkey', 'kernel', 'shim-signed', 'shim',
'psa', 'fail2ban',
'libpam-plesk', 'libaps', 'mod-security-v3',
'libapache2-modsecurity-plesk', 'libapache2-mod-aclr2-psa',
'libapache2-mod-sysenv-psa', 'libapache2-mod-fcgid-psa',
'FortiMonitor Agent',
'Integration Services', 'SSMS Post Install Tasks',
'SAV Dynamic Interface',
'pciutils', 'usbutils', 'dmidecode', 'irqbalance', 'kmod',
'base-files', 'base-passwd', 'hostname', 'filesystem', 'tzdata',
'coreutils', 'util-linux', 'sed', 'grep', 'diffutils', 'findutils',
'tar', 'gzip', 'ca-certificates', 'bash-completion', 'distro-info',
'initscripts', 'chkconfig', 'authselect', 'debconf', 'aptitude',
'jq', 'bc', 'at', 'gd', 'm4', 'zip', 'rpm', 'lua',
])
# Match package names that start with these (true prefix — not substring).
INFRA_PACKAGE_PREFIXES = (
'cpanel-', 'plesk-', 'ea-', 'wp-toolkit-', 'installatron-',
'psa-', 'sw-', 'mod_passenger', 'pp18.',
'libapache2-mod-passenger',
'imunify', 'ai-bolit', 'alt-common',
'Plesk',
'kernel-', 'linux-image-', 'linux-headers-', 'linux-modules-',
'grub-', 'grub2-',
'microcode-', 'firmware-', 'alsa-', 'intel-', 'amd64-', 'base-',
'open-vm-', 'qemu-guest-',
'lib', 'alt', 'perl-', 'ruby-', 'php-', 'glibc', 'glib2', 'nss-',
'openssl', 'gnupg', 'gnutls',
'ubuntu-', 'debian-', 'task-',
'adwaita-', 'hicolor-', 'language-pack-', 'language-selector-',
'fonts-', 'gnome-', 'xserver-xorg-', 'desktop-file-', 'shared-mime-',
'apt-', 'dpkg-',
'redhat-', 'fedora-', 'centos-', 'almalinux-', 'selinux-policy',
'dracut-', 'systemd-', 'dnf-', 'yum-', 'rpm-', 'sendmail',
'Microsoft Visual C++', 'Microsoft .NET', 'Microsoft ASP.NET',
'Microsoft Edge', 'Microsoft ODBC', 'Microsoft SQL Server',
'Microsoft VSS Writer', 'Microsoft Silverlight',
'DirectX', 'WebView2',
'KB', 'Cumulative Update', 'Security Update', 'Service Stack',
'Monthly Rollup', 'Critical Update',
'Windows Driver Package', 'Integration Services',
'Intel(R)', 'Realtek',
)
# Substring match only where the pattern is not a reliable prefix (e.g. *-passenger).
INFRA_PACKAGE_CONTAINS = (
'passenger',
'Critical Update for SQL Server',
'SQL Server',
'FortiMonitor',
)
# Threshold for detecting automated batch installations (system updates, provisioning).
# Based on operational data: human administrators typically install 1-5 packages at a time,
# while dnf/apt updates commonly install 10-170+ packages. The value 8 provides a buffer
# to avoid false positives from legitimate multi-package manual installations (e.g., a user
# installing a development stack like python3, pip, git, vim, htop).
BATCH_INSTALL_THRESHOLD = 8
# Time window in minutes for burst detection. System updates/provisioning scripts
# typically complete within 5-20 minutes. The 30-minute window captures related
# installations (packages with dependencies) while separating distinct manual operations
# hours apart. Using a sliding window with ±15 minutes around each package install time,
# so a 30-minute window means packages within 15 minutes before or after are grouped.
BATCH_INSTALL_TIME_WINDOW_MINUTES = 30
def collect_vm_metrics(self, payload: Dict[str, Any]) -> NydusResult: # noqa: C901
op_name = 'collect_vm_metrics'
start_date = payload.get('start_date', '') or ''
end_date = payload.get('end_date', '')
metric_types = payload.get('metric_types', ['installed_packages'])
if not (start_date and str(start_date).strip()):
return False, self.build_result_dict(
'', 'start_date is required', op_name)
start_date = str(start_date).strip()
start_date = self._normalize_date(start_date)
if start_date is None:
return False, self.build_result_dict(
'', 'Invalid start_date format: {}'.format(
payload.get('start_date', '')), op_name)
if end_date:
end_date = self._normalize_date(end_date)
if end_date is None:
return False, self.build_result_dict(
'', 'Invalid end_date format: {}'.format(
payload.get('end_date', '')), op_name)
if not (end_date and str(end_date).strip()):
end_date = ''
if start_date and end_date:
try:
prov_dt = datetime.strptime(start_date, '%Y-%m-%dT%H:%M:%S')
end_dt = datetime.strptime(end_date, '%Y-%m-%dT%H:%M:%S')
if end_dt < prov_dt:
return False, self.build_result_dict(
'', 'end_date must be equal to or after start_date', op_name)
except ValueError:
pass
collectors = {
'installed_packages': self._collect_installed_packages,
'processes_services_resources': self._collect_processes_services_resources,
}
metrics = {}
errors = []
for metric_type in metric_types:
collector = collectors.get(metric_type)
if collector is None:
errors.append("Unknown metric type: {}".format(metric_type))
continue
try:
metrics[metric_type] = collector(start_date, end_date)
except NotImplementedError:
errors.append("{}: not supported on {}".format(metric_type, str(self)))
except Exception as e:
LOG.exception("Failed to collect metric: %s", metric_type)
errors.append("{}: {}".format(metric_type, str(e)))
result = {
'collected_at': datetime.utcnow().strftime(CANONICAL_TIMESTRING_FORMAT),
'start_date': start_date,
'os_class': str(self),
'metrics': metrics,
}
if end_date:
result['end_date'] = end_date
if errors:
result['errors'] = errors
success = len(metrics) > 0
return success, self.build_result_dict(
json.dumps(result),
'; '.join(errors) if errors else '',
op_name
)
def _collect_installed_packages(self, start_date: str,
end_date: str = '') -> Dict[str, Any]:
raise NotImplementedError
def _collect_processes_services_resources(self, start_date: str,
end_date: str = '') -> Dict[str, Any]:
raise NotImplementedError
@staticmethod
def _normalize_date(start_date: str) -> Union[str, None]:
cleaned = start_date.replace('Z', '').split('.')[0]
for fmt in ('%Y-%m-%dT%H:%M:%S', '%Y-%m-%d'):
try:
dt = datetime.strptime(cleaned, fmt)
return dt.strftime('%Y-%m-%dT%H:%M:%S')
except ValueError:
continue
return None
@staticmethod
def _filter_by_start_date(packages: List[Dict[str, str]],
start_date: str,
end_date: str = '',
buffer_hours: int = 0) -> List[Dict[str, str]]:
packages = [p for p in packages
if p.get('installed_date') and p['installed_date'] != 'unknown']
if start_date:
try:
cutoff = datetime.strptime(start_date, '%Y-%m-%dT%H:%M:%S')
cutoff = cutoff + timedelta(hours=buffer_hours)
cutoff_str = cutoff.strftime('%Y-%m-%dT%H:%M:%S')
packages = [p for p in packages
if p['installed_date'] >= cutoff_str]
except ValueError:
pass
if end_date:
try:
end_str = end_date
if end_str.endswith('T00:00:00'):
end_str = end_str.replace('T00:00:00', 'T23:59:59')
packages = [p for p in packages
if p['installed_date'] <= end_str]
except (ValueError, TypeError):
pass
return packages
@staticmethod
def _parse_install_times(packages: List[Dict[str, str]]):
"""Parse install dates and pair each package with its datetime.
Converts the 'installed_date' string from each package dict into a datetime object.
Packages with missing, 'unknown', or unparsable dates get None as their datetime.
Returns a list sorted by install time, with None-dated packages at the beginning
(treated as datetime.min for sorting purposes).
:param packages: List of package dicts, each expected to have 'installed_date' key
:return: List of (datetime_or_None, package_dict) tuples, sorted by datetime ascending
"""
parsed = []
for pkg in packages:
install_date = pkg.get('installed_date', '')
if not install_date or install_date == 'unknown':
parsed.append((None, pkg))
continue
try:
parsed.append((datetime.strptime(install_date, '%Y-%m-%dT%H:%M:%S'), pkg))
except ValueError:
# Unparsable date format - treat as unknown
parsed.append((None, pkg))
# Sort by datetime, placing None (unknown) dates first
parsed.sort(key=lambda x: x[0] if x[0] else datetime.min)
return parsed
@staticmethod
def _find_burst_indices(parsed_packages, threshold: int, time_window_minutes: int):
"""Detect packages installed in bursts using a sliding window algorithm.
Uses a two-pointer sliding window technique to efficiently find all packages
whose install time falls within a "burst window" - a period where >= threshold
packages were installed within time_window_minutes of each other.
Algorithm: O(n) time complexity
- For each package with a valid timestamp, create a window of ±(time_window_minutes/2)
- Count how many packages fall within that window
- If count >= threshold, mark all packages in the window as part of a burst
Example: threshold=8, time_window_minutes=30
- Package at 10:15 creates window [10:00, 10:30]
- If 8+ packages installed between 10:00-10:30, all are marked as burst
:param parsed_packages: List of (datetime_or_None, package_dict) tuples from _parse_install_times
:param threshold: Minimum number of packages in a window to qualify as a burst
:param time_window_minutes: Total time window in minutes (split ±half around each timestamp)
:return: Set of indices (into parsed_packages) that are part of any burst
"""
burst = set()
half_window = timedelta(minutes=time_window_minutes / 2)
# Extract only packages with valid timestamps, keeping their original indices
# valid_entries: [(original_index_in_parsed_packages, datetime), ...]
valid_entries = [
(i, install_time)
for i, (install_time, _pkg) in enumerate(parsed_packages)
if install_time is not None
]
# Two-pointer sliding window
left = 0
right = 0
total_valid = len(valid_entries)
# For each package, check if it's part of a burst window
for current_pos, (pkg_index, install_time) in enumerate(valid_entries):
window_start = install_time - half_window
window_end = install_time + half_window
# Move left pointer to the start of the window
while left < total_valid and valid_entries[left][1] < window_start:
left += 1
# Move right pointer to the end of the window (inclusive)
while right < total_valid and valid_entries[right][1] <= window_end:
right += 1
# If window contains >= threshold packages, mark them all as burst
if right - left >= threshold:
burst.update(pkg_idx for pkg_idx, _time in valid_entries[left:right])
return burst
@staticmethod
def _filter_batch_installs(packages: List[Dict[str, str]],
threshold: int = 8,
time_window_minutes: int = 30) -> List[Dict[str, str]]:
"""Exclude packages installed in large batches within a short time window.
Filters out packages that were likely installed by automated system updates or
provisioning scripts, keeping only packages that appear to be manually installed.
Rationale:
- System updates (dnf/apt/yum) typically install 10-170+ packages within 5-20 minutes
- Human administrators typically install 1-5 packages at a time with longer gaps
- Threshold of 8 provides buffer to avoid false positives on legitimate multi-package installs
:param packages: List of package dicts with 'installed_date' field
:param threshold: Minimum packages in time window to consider a "burst" (default: 8)
:param time_window_minutes: Sliding window size in minutes (default: 30)
:return: Filtered list containing only packages NOT part of detected bursts
"""
parsed_packages = Ops._parse_install_times(packages)
burst_indices = Ops._find_burst_indices(parsed_packages, threshold, time_window_minutes)
# Return packages whose index is NOT in the burst set
return [pkg for i, (_datetime, pkg) in enumerate(parsed_packages) if i not in burst_indices]
@staticmethod
def _package_installed_after(pkg: Dict[str, str], cutoff: datetime) -> bool:
"""True if pkg installed_date parses and is strictly after cutoff."""
raw = (pkg.get('installed_date') or '')
try:
return datetime.strptime(raw, '%Y-%m-%dT%H:%M:%S') > cutoff
except (ValueError, TypeError):
return False
@staticmethod
def _parse_package_lines(raw_output: str) -> List[Dict[str, str]]:
packages = []
if not raw_output or not raw_output.strip():
return packages
for line in raw_output.strip().split('\n'):
parts = line.split('|')
if len(parts) != 4:
continue
packages.append({
'name': parts[0].strip(),
'version': parts[1].strip(),
'installed_date': parts[2].strip(),
'source': parts[3].strip(),
})
return packages
def _build_package_result(self, packages: List[Dict[str, str]]) -> Dict[str, Any]:
def _strip_arch(name):
if ':' in name:
return name.split(':')[0]
return name
def _is_infra(name):
if not name:
return True
base = _strip_arch(name)
if base in self.INFRA_PACKAGE_EXCLUSIONS:
return True
for prefix in self.INFRA_PACKAGE_PREFIXES:
if base.startswith(prefix):
return True
for needle in self.INFRA_PACKAGE_CONTAINS:
if needle in base:
return True
return False
packages = [p for p in packages if not _is_infra(p.get('name'))]
truncated = False
if len(packages) > self.MAX_METRIC_ITEMS:
packages = packages[:self.MAX_METRIC_ITEMS]
truncated = True
result = {
'count': len(packages),
'items': packages,
}
if truncated:
result['truncated'] = True
json_str = json.dumps(result)
while len(json_str) > self.MAX_METRIC_JSON_SIZE and packages:
packages = packages[:len(packages) // 2]
result['items'] = packages
result['count'] = len(packages)
result['truncated'] = True
json_str = json.dumps(result)
return result
def _result_handler(self, exit_code: int, outs: str, errs: str, op_name: str,
intermediate_result: Dict[str, Any] = None) -> NydusResult:
"""Take the result from a run command and check for retryable errors.
If found, return a Retry. If not, return a Nydus result tuple.
:param exit_code: The exit code from the executed command
:param outs: The stdout output from the executed command
:param errs: The stderr output from the executed command
:param op_name: The name of the op
:param intermediate_result: Dict containing metadata for retries
:return: A Nydus result tuple, or Retry
"""
LOG.info(self.RETRYABLE_ERRS)
success, result = self.build_result_from_cmd_output(exit_code, outs, errs, op_name, op_name + " succeeded")
if not success:
for retryable_err in self.RETRYABLE_ERRS:
if retryable_err in errs:
if intermediate_result is None:
intermediate_result = {}
intermediate_result['retryable_error'] = retryable_err
return Retry(intermediate_result)
return success, result