ipfixprobed: fix DPDK parser

fix queue size and workers affinity

Author: Pavel Siska

init/config2args.py

@@ -7,6 +7,7 @@
 import json
 import re
 from pathlib import Path
+from typing import List
 
 def load_config(file_path):
     try:
@@ -54,7 +55,7 @@ def process_input_plugin(config):
 
     return " ".join(params)
 
-def get_cpus_for_pci_device(pci_address: str) -> list[int]:
+def get_cpus_for_pci_device(pci_address: str) -> List[int]:
     """
     Gets the list of CPU IDs associated with the NUMA node corresponding to the given PCI address.
 
@@ -63,15 +64,14 @@ def get_cpus_for_pci_device(pci_address: str) -> list[int]:
     """
     # Get the NUMA node
     numa_path = Path(f"/sys/bus/pci/devices/{pci_address}/numa_node")
-    if not numa_path.exists():
-        raise FileNotFoundError(f"NUMA node info for PCI address {pci_address} does not exist.")
-
-    numa_node = numa_path.read_text().strip()
-    if numa_node == "-1":
-        raise ValueError(f"Device {pci_address} is not assigned to any NUMA node.")
+    numa_node = 0
+    if numa_path.exists():
+        numa_node = numa_path.read_text().strip()
+        if numa_node == "-1":
+            numa_node = 0
 
     # Run lscpu to get CPU information
-    result = subprocess.run(["lscpu"], capture_output=True, text=True, check=True)
+    result = subprocess.run(["lscpu"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, check=True)
     lines = result.stdout.splitlines()
 
     # Find the line corresponding to the NUMA node
@@ -84,7 +84,7 @@ def get_cpus_for_pci_device(pci_address: str) -> list[int]:
 
     raise RuntimeError(f"Could not find CPU list for NUMA node {numa_node}.")
 
-def parse_cpu_list(cpu_list_str: str) -> list[int]:
+def parse_cpu_list(cpu_list_str: str) -> List[int]:
     """
     Converts a CPU range string like "1,3,5-7" to a list of individual CPU numbers [1, 3, 5, 6, 7].
 
@@ -138,11 +138,18 @@ def process_input_dpdk_plugin(settings):
         workers_cpu_list = cpu_list[:rx_queues]
 
     # Main parameter for DPDK with $eal_opts
-    primary_param = f"-i \"dpdk;p={','.join(str(i) for i in range(nic_count))};"
+    first_cpu = workers_cpu_list[0]
+    if first_cpu is not None:
+        primary_param = f"-i \"dpdk@{first_cpu};p={','.join(str(i) for i in range(nic_count))};"
+    else:
+        primary_param = f"-i \"dpdk;p={','.join(str(i) for i in range(nic_count))};"
+
     burst_size = settings.get("burst_size", 64)
     if burst_size is not None:
         primary_param += f"b={burst_size};"
 
+    primary_param += f"q={rx_queues};"
+
     mempool_size = settings.get("mempool_size", 8192)
     if mempool_size is not None:
         primary_param += f"m={mempool_size};"
@@ -153,11 +160,7 @@ def process_input_dpdk_plugin(settings):
     primary_param += f"eal={eal}\""
 
     params = []
-    first_cpu = workers_cpu_list[0]
-    if first_cpu is not None:
-        params.append(f"{primary_param}@{first_cpu}")
-    else:
-        params.append(primary_param)
+    params.append(primary_param)
 
     for i in range(1, rx_queues):
         cpu = workers_cpu_list[i]