fix: parallelize feature is moved to source block (#2536)

The new PR we saw improved way of ingestion +--------------------------------------------------------+------------+-------------------+ | Configuration | %Cpu(s) us | Average rate (msg/sec) | +--------------------------------------------------------+------------+-------------------+ | SC4S main parallelize off --active-connections=1 | 9.0 | 14144.10 | | SC4S main parallelize off --active-connections=10 | 59.3 | 73743.32 | | SC4S old parallelize on (10 threads) --active-connections=1 | 11.8 | 17184.52 | | SC4S pr-2536 parallelize on (10 threads) --active-connections=1 | 58.4 | 77842.18 | +--------------------------------------------------------+------------+-------------------+
splunk · Jul 24, 2024 · 16fb3a5 · 16fb3a5
1 parent 1f651ae
commit 16fb3a5
Show file tree

Hide file tree

Showing 4 changed files with 5 additions and 6 deletions.
diff --git a/package/etc/conf.d/log_paths/2/lp_dest_alts_global/plugin.jinja b/package/etc/conf.d/log_paths/2/lp_dest_alts_global/plugin.jinja
@@ -1,8 +1,5 @@
 
 log{
-    {%- if enable_parallelize == True %}
-    parallelize(partitions({{ parallelize_no_partitions }}));
-    {%- endif %}
     {%- if mode == "SELECT" %}
     parser {
         app-parser(topic(sc4s-lp-dest-select-{{ destination }}));

diff --git a/package/etc/conf.d/log_paths/2/lp_dest_alts_global/plugin.py b/package/etc/conf.d/log_paths/2/lp_dest_alts_global/plugin.py
@@ -87,7 +87,5 @@ def normalize_env_variable_input(env_variable: str):
         mode=m["mode"],
         filter=m["filter"],
         dtype=m["dtype"],
-        enable_parallelize=normalize_env_variable_input(f"SC4S_ENABLE_PARALLELIZE"),
-        parallelize_no_partitions=int(os.getenv(f"SC4S_PARALLELIZE_NO_PARTITION", 4)),
     )
     print(msg)
diff --git a/package/etc/conf.d/sources/source_syslog/plugin.jinja b/package/etc/conf.d/sources/source_syslog/plugin.jinja
@@ -103,7 +103,9 @@ source s_{{ port_id }} {
             {%- endif %}
             {%- endfor %}
         };
-
+        {%- if enable_parallelize == True %}
+        parallelize(partitions({{ parallelize_no_partitions }}));
+        {%- endif %}
         {%- if vendor and product %}
         parser {
             p_set_netsource_fields(

diff --git a/package/etc/conf.d/sources/source_syslog/plugin.py b/package/etc/conf.d/sources/source_syslog/plugin.py
@@ -131,5 +131,7 @@ def normalize_env_variable_input(env_variable: str):
             "HIGH:!aNULL:!eNULL:!kECDH:!aDH:!RC4:!3DES:!CAMELLIA:!MD5:!PSK:!SRP:!KRB5:@STRENGTH",
         ),
         ebpf_no_sockets=int(os.getenv("SC4S_EBPF_NO_SOCKETS", 4)),
+        enable_parallelize=normalize_env_variable_input(f"SC4S_ENABLE_PARALLELIZE"),
+        parallelize_no_partitions=int(os.getenv(f"SC4S_PARALLELIZE_NO_PARTITION", 4)),
     )
     print(outputText)