sd

Author: Pavel Siska

init/config2args.py

@@ -68,6 +68,18 @@ def process_input_dpdk_plugin(settings):
     if eal_opts:
         eal += f" {eal_opts}"
 
+    workers_cpu_list = settings.get("workers_cpu_list")
+    if workers_cpu_list:
+        if isinstance(workers_cpu_list, str):
+            workers_cpu_list = [cpu.strip() for cpu in workers_cpu_list.split(",")]
+        elif not isinstance(workers_cpu_list, list):
+            raise ValueError("workers_cpu_list must be a list or a comma-separated string")
+
+        if len(workers_cpu_list) != rx_queues:
+            raise ValueError("The number of CPUs in workers_cpu_list must match the number of RX queues")
+    else:
+        workers_cpu_list = [None] * rx_queues
+
     # Main parameter for DPDK with $eal_opts
     primary_param = f"-i \"dpdk;p={','.join(str(i) for i in range(nic_count))};"
     burst_size = settings.get("burst_size", 64)
@@ -83,7 +95,19 @@ def process_input_dpdk_plugin(settings):
         primary_param += f"mtu={mtu};"
     primary_param += f"eal={eal}\""
 
-    params = [primary_param] + [f"-i dpdk" for _ in range(rx_queues - 1)]
+    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)
+
+    for i in range(1, rx_queues):
+        cpu = workers_cpu_list[i]
+        if cpu is not None:
+            params.append(f"-i dpdk@{cpu}")
+        else:
+            params.append(f"-i dpdk")
 
     return " ".join(params)
 
@@ -285,11 +309,21 @@ def process_output_ipfix_plugin(settings):
         return f'{";".join(params)}"'
 
     # Main parameters
-    host = settings.get("collector", {}).get("host")
-    port = settings.get("collector", {}).get("port")
-    mtu = settings.get("mtu")
-    exporter_id = settings.get("exporter", {}).get("id")
-    exporter_dir = settings.get("exporter", {}).get("dir")
+    collector = settings.get("collector")
+    if collector is None:
+        raise ValueError("collector must be specified in the ipfix plugin configuration.")
+
+    host = collector.get("host")
+    if host is None:
+        raise ValueError("host must be specified in the ipfix (collector) configuration. ")
+
+    port = collector.get("port")
+    if port is None:
+        raise ValueError("port must be specified in the ipfix (collector) configuration. ")
+
+    mtu = settings.get("mtu", 1518)
+    exporter_id = settings.get("exporter", {}).get("id", 1)
+    exporter_dir = settings.get("exporter", {}).get("dir", 0)
 
     if host is not None:
         params.append(f"host={host}")
@@ -311,24 +345,36 @@ def process_output_ipfix_plugin(settings):
     is_tcp = False
     if "udp" in protocol:
         params.append("udp")
-        template_refresh = protocol["udp"].get("template_refresh")
-        if template_refresh is not None:
+        udp = protocol.get("udp")
+        if udp is not None:
+            template_refresh = udp.get("template_refresh", 600)
+            if template_refresh is not None:
+                params.append(f"template={template_refresh}")
+        else:
+            template_refresh = 600
             params.append(f"template={template_refresh}")
     elif "tcp" in protocol:
         is_tcp = True
-        non_blocking = protocol["tcp"].get("non_blocking")
-        if non_blocking is not None:
-            params.append("non-blocking-tcp")
+        tcp = protocol.get("tcp")
+        if tcp is not None:
+            non_blocking = tcp.get("non_blocking", {})
+            if non_blocking is not None:
+                params.append("non-blocking-tcp")
+    else:
+        raise ValueError("Invalid options for ipfix protocol. Must be either 'udp' or 'tcp'.")
 
     # LZ4 compression (only valid with TCP)
-    compression = settings.get("compression", {}).get("lz4", {})
-    if compression.get("enabled"):
-        if not is_tcp:
-            raise ValueError("LZ4 compression can only be used with TCP.")
-        params.append("lz4-compression")
-        buffer_size = compression.get("buffer_size")
-        if buffer_size is not None:
-            params.append(f"lz4-buffer-size={buffer_size}")
+    compression = settings.get("compression", {})
+    if compression is not None:
+        lz4 = compression.get("lz4", {})
+        if lz4 is not None:
+            if lz4.get("enabled"):
+                if not is_tcp:
+                    raise ValueError("LZ4 compression can only be used with TCP.")
+                params.append("lz4-compression")
+                buffer_size = lz4.get("buffer_size")
+                if buffer_size is not None:
+                    params.append(f"lz4-buffer-size={buffer_size}")
 
     return f'{";".join(params)}"'