T7424: Refactor resource validation and broaden cases (#38)

* T7424: Refactor and extend resource usage verification on commit for VPP CLI

T7424: Fix ruff errors

* T7424: Implement check for smoke tests runtime; reduce resource requirements for test environments

T7424: Fix errors in calculating the skipped and reserved CPU cores; Adjust default main heap size value.

* T7424: Refactor the CPU checks logic; Add total CPU usage check

T7424: Fix CPU reserve and skip cores calculations; Add total CPU usage check

T7424: Refactor smoketests to reflect new logic

* T7424: Refactor the CPU and memory checks logic

---------

Co-authored-by: oniko94 <onikolaiev94@outlook.com>

5 files changed, 518 insertions, 0 deletions

diff --git a/python/vyos/vpp/config_resource_checks/__init__.py b/python/vyos/vpp/config_resource_checks/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/python/vyos/vpp/config_resource_checks/__init__.py
diff --git a/python/vyos/vpp/config_resource_checks/cpu.py b/python/vyos/vpp/config_resource_checks/cpu.py
new file mode 100644
index 000000000..6f360ae74
--- /dev/null
+++ b/python/vyos/vpp/config_resource_checks/cpu.py
@@ -0,0 +1,76 @@
+# Used for validating estimated CPU/physical cores use
+#
+# Copyright (C) 2025 VyOS Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+from vyos.utils.cpu import get_available_cpus, get_core_count
+
+from vyos.vpp.config_resource_checks.resource_defaults import get_resource_defaults
+
+
+# Get default value for reserved cpu cores
+reserved_cpus = get_resource_defaults().get('reserved_cpu_cores')
+
+
+def available_cores_count(cpu_settings: dict) -> int:
+    core_count = get_core_count()
+
+    if cpu_settings.get('main_core'):
+        core_count -= 1
+
+    skip_cores = int(cpu_settings.get('skip_cores', 0))
+    # The default settings assume that
+    # at least 2 CPU cores should remain reserved for system use
+    # (only in case of current runtime is not smoke test)
+    if skip_cores < reserved_cpus:
+        core_count -= reserved_cpus
+    else:
+        core_count -= skip_cores
+
+    return core_count
+
+
+def available_cores_list(skip_cores: int) -> list:
+    # Available cores are all CPU cores without first N skipped cores that will not be used
+    # Get all available physical cores - use set to filter out unique values
+    cpu_cores = set(map(lambda el: el['cpu'], get_available_cpus()))
+    cpu_cores = list(cpu_cores)
+
+    return cpu_cores[skip_cores:]
+
+
+def worker_cores_list(iface: str, worker_ranges: list) -> list:
+    all_core_numbers = []
+    for worker_range in worker_ranges:
+        core_numbers = worker_range.split('-')
+
+        if int(core_numbers[0]) > int(core_numbers[-1]):
+            raise ValueError(
+                f'Range for "{iface} workers {worker_range}" is not correct'
+            )
+
+        all_core_numbers.extend(range(int(core_numbers[0]), int(core_numbers[-1]) + 1))
+
+    # Check for duplicates
+    duplicates = set(
+        [str(x) for n, x in enumerate(all_core_numbers) if x in all_core_numbers[:n]]
+    )
+    if duplicates:
+        raise ValueError(
+            f'Some workers in "{iface} workers" are duplicated: #{",".join(list(duplicates))}'
+        )
+
+    return all_core_numbers
diff --git a/python/vyos/vpp/config_resource_checks/memory.py b/python/vyos/vpp/config_resource_checks/memory.py
new file mode 100644
index 000000000..44f0c9358
--- /dev/null
+++ b/python/vyos/vpp/config_resource_checks/memory.py
@@ -0,0 +1,152 @@
+# Used for memory consumption calculations
+#
+# Copyright (C) 2025 VyOS Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+import re
+
+from vyos.utils.process import cmd
+from vyos.vpp.utils import (
+    human_memory_to_bytes,
+    human_page_memory_to_bytes,
+)
+from vyos.vpp.config_resource_checks.resource_defaults import get_resource_defaults
+
+
+# Get default values for resource checks
+defaults = get_resource_defaults()
+
+
+def get_total_hugepages_free_memory() -> int:
+    """
+    Returns the total amount of hugepage-backed free memory (in bytes)
+    as reported by /proc/meminfo
+    """
+    info = {}
+    with open('/proc/meminfo', 'r') as meminfo:
+        for line in meminfo:
+            if line.startswith('Huge'):
+                key, value, *_ = line.strip().split()
+                info[key.rstrip(':')] = int(value)
+
+    hugepages_free = info.get('HugePages_Free')
+    hugepage_size = info.get('Hugepagesize') * 1024
+
+    return hugepage_size * hugepages_free
+
+
+def get_numa_count():
+    """
+    Run `numactl --hardware` and parse the 'available:' line.
+    """
+    out = cmd('numactl --hardware')
+    # e.g. "available: 2 nodes (0-1)"
+    m = re.search(r'available:\s*(\d+)\s+nodes', out)
+    return int(m.group(1)) if m else 0
+
+
+def get_memory_from_kernel_settings(settings: dict) -> int:
+    hugepage_settings = settings.get('hugepage_size', {})
+
+    total_bytes = 0
+    for size_str, info in hugepage_settings.items():
+        count = int(info.get('hugepage_count', 0))
+        page_bytes = human_memory_to_bytes(size_str)
+        total_bytes += count * page_bytes
+
+    return total_bytes
+
+
+def buffer_size(settings: dict) -> int:
+    numa_count = get_numa_count()
+    buffers_per_numa = int(
+        settings.get('buffers', {}).get(
+            'buffers_per_numa', defaults.get('buffers_per_numa')
+        )
+    )
+    data_size = int(
+        settings.get('buffers', {}).get('data_size', defaults.get('data_size'))
+    )
+    buffers_memory = buffers_per_numa * data_size * numa_count
+    return buffers_memory
+
+
+def main_heap_page_size(settings: dict) -> int:
+    heap_page_size = settings.get('memory', {}).get(
+        'main_heap_page_size', defaults.get('main_heap_page_size')
+    )
+    return human_page_memory_to_bytes(heap_page_size)
+
+
+def memory_main_heap(settings: dict) -> int:
+    heap_size = settings.get('memory', {}).get(
+        'main_heap_size', defaults.get('main_heap_size')
+    )
+    return human_memory_to_bytes(heap_size)
+
+
+def ipv6_heap_size(settings: dict) -> int:
+    heap_size = settings.get('ipv6', {}).get(
+        'heap_size', defaults.get('ipv6_heap_size')
+    )
+    return human_memory_to_bytes(heap_size)
+
+
+def total_heap_size(heap_size: int, heap_page_size: int) -> int:
+    return (heap_size + heap_page_size - 1) & ~(heap_page_size - 1)
+
+
+def statseg_size(settings: dict) -> int:
+    statseg_memory = settings.get('statseg', {}).get(
+        'size', defaults.get('statseg_heap_size')
+    )
+    return human_memory_to_bytes(statseg_memory)
+
+
+def statseg_page_size(settings: dict) -> int:
+    page_size = settings.get('statseg', {}).get('page_size', 'default')
+    return human_page_memory_to_bytes(page_size)
+
+
+def total_statseg_size(_statseg_size: int, _statseg_page: int) -> int:
+    return (_statseg_size + _statseg_page - 1) & ~(_statseg_page - 1)
+
+
+def total_memory_required(settings: dict) -> int:
+    mem_required = 0
+
+    mem_stats = {
+        'memory_buffers': buffer_size(settings),
+        'netlink_buffer_size': int(
+            settings.get('lcp', {}).get(
+                'rx_buffer_size', defaults.get('netlink_rx_buffer_size')
+            )
+        ),
+        'heap_size': total_heap_size(
+            heap_size=memory_main_heap(settings),
+            heap_page_size=main_heap_page_size(settings),
+        ),
+        'statseg_size': total_statseg_size(
+            _statseg_size=statseg_size(settings),
+            _statseg_page=statseg_page_size(settings),
+        ),
+        'ipv6_heap_size': ipv6_heap_size(settings),
+    }
+
+    for stat in mem_stats:
+        mem_required += mem_stats[stat]
+
+    return mem_required
diff --git a/python/vyos/vpp/config_resource_checks/resource_defaults.py b/python/vyos/vpp/config_resource_checks/resource_defaults.py
new file mode 100644
index 000000000..3c618accd
--- /dev/null
+++ b/python/vyos/vpp/config_resource_checks/resource_defaults.py
@@ -0,0 +1,62 @@
+# Default values for resource consumption checks
+#
+# Copyright (C) 2025 VyOS Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+import copy
+import os
+
+
+__default_resource_map = {
+    # Default amount of buffers per NUMA (populated CPU socket)
+    'buffers_per_numa': 16384,
+    # Default size of buffer (in bytes)
+    'data_size': 2048,
+    # Default hugepage size for VPP
+    'hugepage_size': '2M',
+    # Default amount of memory allocated for VPP exclusive usage
+    'main_heap_size': '3G',
+    # Default main heap page size
+    'main_heap_page_size': '2M',
+    # Default size of buffers transferred via netlink
+    'netlink_rx_buffer_size': 212992,
+    # Default amount of memory allocated for VPP stats segment usage
+    'statseg_heap_size': '96M',
+    # Minimal amount of memory required to start VPP
+    'min_memory': '8G',
+    # Minimal number of physical CPU cores required to start VPP
+    'min_cpus': 4,
+    # Reserve at least 2 physical cores
+    'reserved_cpu_cores': 2,
+    # Default heap size for IPv6
+    'ipv6_heap_size': '32M',
+}
+
+
+def get_resource_defaults() -> dict:
+    resource_map = copy.deepcopy(__default_resource_map)
+    # Check if current runtime is smoke tests
+    # Since CI/CD runners are limited in resources, reduce the checks for tests
+    if is_smoketest():
+        resource_map.update(min_memory='6G', min_cpus=2, reserved_cpu_cores=0)
+
+    return resource_map
+
+
+def is_smoketest():
+    if os.path.exists('/tmp/vyos.smoketests.hint'):
+        return True
+    return False
diff --git a/python/vyos/vpp/config_verify.py b/python/vyos/vpp/config_verify.py
index 830d4af73..9cfbfb7cd 100644
--- a/python/vyos/vpp/config_verify.py
+++ b/python/vyos/vpp/config_verify.py
@@ -16,9 +16,19 @@
 # with this program; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
+import psutil
+
 from vyos import ConfigError
+from vyos.utils.cpu import get_core_count as total_core_count
 
 from vyos.vpp.control_host import get_eth_driver
+from vyos.vpp.config_resource_checks import cpu as cpu_checks, memory as mem_checks
+from vyos.vpp.config_resource_checks import resource_defaults
+from vyos.vpp.utils import human_memory_to_bytes, bytes_to_human_memory
+
+
+# Get default values for resource checks
+defaults = resource_defaults.get_resource_defaults()
 
 
 def verify_vpp_remove_kernel_interface(config: dict):
@@ -171,3 +181,221 @@ def verify_dev_driver(iface_name: str, driver_type: str) -> bool:
         raise ConfigError(f'"Driver type {driver_type} is wrong')
 
     return False
+
+
+def verify_vpp_minimum_cpus():
+    """
+    Verify that the host system has enough physical CPU cores
+    Current minimal requirement is 4
+    """
+    min_cpus = defaults.get('min_cpus')
+    if total_core_count() < min_cpus:
+        raise ConfigError(
+            'This system does not meet minimal requirements for VPP. '
+            f'Minimum {min_cpus} CPU cores are required.'
+        )
+
+
+def verify_vpp_minimum_memory():
+    """
+    Verify that the host system has enough RAM
+    Calculate by retrieving the amount of physical memory
+    And the minimal requirement (currently 8 GB). Round before comparing -
+    To avoid situations like when a machine nominally has 8192 MB (8 giga/gibibytes)
+    But the OS sees only 7.75 GB, creating a fail condition for this check
+    """
+    min_mem = defaults.get('min_memory')
+    total_memory = round(psutil.virtual_memory().total / (1024**3))
+    min_memory = round(human_memory_to_bytes(min_mem) / (1024**3))
+
+    if total_memory < min_memory:
+        raise ConfigError(
+            'This system does not meet minimal requirements for VPP. '
+            f'Minimum {min_memory} GB of RAM are required.'
+        )
+
+
+def verify_vpp_memory(config: dict):
+    main_heap_size = mem_checks.memory_main_heap(config['settings'])
+    main_heap_page_size = mem_checks.main_heap_page_size(config['settings'])
+
+    if main_heap_size < 51 << 20:
+        raise ConfigError('The main heap size must be greater than or equal to 51M')
+
+    readable_heap_page = bytes_to_human_memory(main_heap_page_size, 'K')
+
+    if main_heap_page_size > main_heap_size:
+        raise ConfigError(
+            f'The main heap size must be greater than or equal to page-size ({readable_heap_page})'
+        )
+
+    # Get available HupePage memory to compare with required memory for VPP
+    # (if it's smketests environment get system kernel settings for HugePages)
+    if not resource_defaults.is_smoketest():
+        available_memory = mem_checks.get_total_hugepages_free_memory()
+    else:
+        available_memory = mem_checks.get_memory_from_kernel_settings(
+            config['kernel_memory_settings']
+        )
+
+    memory_required = mem_checks.total_memory_required(config['settings'])
+
+    # Check if there is a config currently active
+    # If yes, calculate how much memory it consumes
+    # and exclude it from required memory
+    if config.get('effective'):
+        memory_used = mem_checks.total_memory_required(config['effective']['settings'])
+        # If we want to reduce memory configs then there is nothing to check
+        if memory_used > memory_required:
+            return
+        memory_required -= memory_used
+
+    if memory_required > available_memory:
+        raise ConfigError(
+            'Not enough free memory to start VPP: '
+            f'available: {round(available_memory / 1024 ** 3, 1)} GB, '
+            f'required: {round(memory_required / 1024 ** 3, 1)} GB. '
+            'Please add kernel memory options for HugePages and reboot'
+        )
+
+
+def verify_vpp_settings_cpu_skip_cores(skip_cores: int):
+    cpu_cores = total_core_count()
+
+    # The number of skipped cores must not be greater than
+    # available CPU cores in the system - 1 for main thread
+    if skip_cores > (cpu_cores - 1):
+        raise ConfigError(
+            f'The system does not have enough available CPUs to skip '
+            f'(reduce "cpu skip-cores" to {cpu_cores} or less)'
+        )
+
+
+def verify_vpp_settings_cpu_and_corelist_workers(settings: dict):
+    """
+    `set vpp settings cpu workers` and `set vpp settings cpu corelist-workers`
+    are mutually exclusive!
+    """
+    if (
+        'corelist_workers' in settings or 'workers' in settings
+    ) and 'main_core' not in settings:
+        raise ConfigError('"cpu main-core" is required but not set!')
+
+    if 'corelist_workers' in settings and 'workers' in settings:
+        raise ConfigError(
+            '"cpu corelist-workers" and "cpu workers" cannot be used at the same time!'
+        )
+
+
+def verify_vpp_cpu_main_core(cpu_settings: dict) -> None:
+    """Check that the main core is available"""
+    skip_cores = int(cpu_settings.get('skip_cores', 0))
+    available_cores = cpu_checks.available_cores_list(skip_cores)
+    main_core = int(cpu_settings['main_core'])
+
+    if main_core not in available_cores:
+        raise ConfigError(
+            'Cannot set main core for VPP process: '
+            f'CPU#{main_core} is not available.'
+        )
+
+
+def verify_vpp_settings_cpu_workers(cpu_settings: dict) -> int:
+    """
+    Verify that the system has enough available CPU cores
+    to run a given amount of worker processes (1 worker/core)
+    """
+    workers = int(cpu_settings.get('workers', 0))
+    available_cores = cpu_checks.available_cores_count(cpu_settings)
+
+    if workers > available_cores:
+        raise ConfigError(
+            f'Not enough free CPU cores for {workers} VPP workers '
+            f'(reduce to {available_cores} or less)'
+        )
+
+    return workers
+
+
+def verify_vpp_settings_cpu_corelist_workers(cpu_settings: dict) -> int:
+    """
+    Verify that the CPU cores provided to the config are free and can be used by VPP
+    """
+    workers = cpu_settings.get('corelist_workers')
+    main_core = int(cpu_settings.get('main_core'))
+    skip_cores = int(cpu_settings.get('skip_cores', 0))
+    available_cores = cpu_checks.available_cores_list(skip_cores)
+    try:
+        all_core_nums = cpu_checks.worker_cores_list(
+            iface='cpu corelist', worker_ranges=workers
+        )
+    except ValueError as e:
+        raise ConfigError(str(e))
+
+    error_msg = 'Cannot set VPP "cpu corelist-workers"'
+
+    if main_core in all_core_nums:
+        raise ConfigError(
+            f'CPU#{main_core} is set as main core and should not '
+            'be included to the corelist-workers'
+        )
+
+    invalid_cores = [str(el) for el in all_core_nums if el not in available_cores]
+    if invalid_cores:
+        raise ConfigError(
+            f'{error_msg}: CPU# {",".join(invalid_cores)} are not available.'
+        )
+
+    if len(all_core_nums) > cpu_checks.available_cores_count(cpu_settings):
+        raise ConfigError(f'{error_msg}: Not enough free CPUs in the system.')
+
+    return len(all_core_nums)
+
+
+def verify_vpp_nat44_workers(workers: int, nat44_workers: list):
+    if workers < 1:
+        raise ConfigError(
+            '"nat44 workers" requires cpu workers or corelist-workers to be set!'
+        )
+    try:
+        nat_workers = cpu_checks.worker_cores_list(
+            iface='nat44', worker_ranges=nat44_workers
+        )
+    except ValueError as e:
+        raise ConfigError(str(e))
+
+    invalid_workers = [str(el) for el in nat_workers if el not in range(workers)]
+    if invalid_workers:
+        raise ConfigError(
+            f'Cannot set VPP "nat44 workers": worker(s) #{",".join(invalid_workers)} not available. '
+            f'Available worker ids: {",".join(map(str, range(workers)))}'
+        )
+
+
+def verify_vpp_statseg_size(settings: dict):
+    statseg_size = mem_checks.statseg_size(settings)
+
+    if 'size' in settings.get('statseg'):
+        if statseg_size < 1 << 20:
+            raise ConfigError('The statseg size must be greater than or equal to 1M')
+
+    if 'page_size' in settings['statseg']:
+        statseg_page_size = mem_checks.statseg_page_size(settings)
+        if statseg_page_size > statseg_size:
+            readable_statseg_page = bytes_to_human_memory(statseg_page_size, 'K')
+            raise ConfigError(
+                f'The statseg size must be greater than or equal to page-size ({readable_statseg_page})'
+            )
+
+
+def verify_vpp_interfaces_dpdk_num_queues(qtype: str, num_queues: int, workers: int):
+    """
+    Verify that VPP has enough workers to run the given amount of RX/TX queues
+    1 queue per 1 worker is assumed as default
+    """
+
+    if num_queues > workers:
+        raise ConfigError(
+            f'The number of {qtype} queues cannot be greater than the number of configured VPP workers: '
+            f'workers: {workers}, queues: {num_queues}'
+        )