Allow tuning multiple Logstash pipelines in SOC

Add logstash:pipeline_settings, a per-pipeline YAML block rendered into
pipelines.yml. Unset settings fall back to logstash.yml. Restart logstash
when pipelines.yml changes, and add the missing managerhype annotation.

Fixes #15090

salt/logstash/config.sls

@@ -81,6 +81,14 @@ ls_custom_pipeline_conf_{{assigned_pipeline}}_{{pipeline}}:
 
 
 {% for assigned_pipeline in ASSIGNED_PIPELINES %}
+{# per-pipeline overrides from pipelines.yml win over the global logstash.yml values #}
+{% set PARSED_OVERRIDES = (LOGSTASH_MERGED.get('pipeline_settings', {}).get(assigned_pipeline, '') | load_yaml) or {} %}
+{% if PARSED_OVERRIDES is not mapping %}
+{%   do salt.log.warning('logstash: ignoring malformed pipeline_settings for pipeline ' ~ assigned_pipeline ~ '; expected "setting: value" pairs') %}
+{% endif %}
+{% set PIPELINE_OVERRIDES = PARSED_OVERRIDES if PARSED_OVERRIDES is mapping else {} %}
+{% set THREADS = PIPELINE_OVERRIDES.get('pipeline.workers', LOGSTASH_MERGED.config.pipeline_x_workers) %}
+{% set BATCH = PIPELINE_OVERRIDES.get('pipeline.batch.size', LOGSTASH_MERGED.config.pipeline_x_batch_x_size) %}
     {% for CONFIGFILE in LOGSTASH_MERGED.defined_pipelines[assigned_pipeline] %}
 ls_pipeline_{{assigned_pipeline}}_{{CONFIGFILE.split('.')[0] | replace("/","_") }}:
   file.managed:
@@ -92,8 +100,8 @@ ls_pipeline_{{assigned_pipeline}}_{{CONFIGFILE.split('.')[0] | replace("/","_")
         GLOBALS: {{ GLOBALS }}
         ES_USER: "{{ salt['pillar.get']('elasticsearch:auth:users:so_elastic_user:user', '') }}"
         ES_PASS: "{{ salt['pillar.get']('elasticsearch:auth:users:so_elastic_user:pass', '') }}"
-        THREADS: {{ LOGSTASH_MERGED.config.pipeline_x_workers }}
-        BATCH: {{ LOGSTASH_MERGED.config.pipeline_x_batch_x_size }}
+        THREADS: {{ THREADS }}
+        BATCH: {{ BATCH }}
       {% else %}
     - name: /opt/so/conf/logstash/pipelines/{{assigned_pipeline}}/{{CONFIGFILE.split('/')[1]}}
       {% endif %}

salt/logstash/defaults.yaml

@@ -60,6 +60,16 @@ logstash:
     custom008: PLACEHOLDER
     custom009: PLACEHOLDER
     custom010: PLACEHOLDER
+  pipeline_settings:
+    fleet: ''
+    manager: ''
+    receiver: ''
+    search: ''
+    custom0: ''
+    custom1: ''
+    custom2: ''
+    custom3: ''
+    custom4: ''
   settings:
     lsheap: 500m
   config:

salt/logstash/enabled.sls

@@ -105,6 +105,7 @@ so-logstash:
     {% endif %}
     - watch:
       - file: lsetcsync
+      - file: lspipelinesyml
       - file: trusttheca
       {% if GLOBALS.is_manager %}
       - file: elasticsearch_cacerts

salt/logstash/etc/pipelines.yml.jinja

@@ -1,4 +1,17 @@
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
+{%- set PIPELINE_SETTINGS = LOGSTASH_MERGED.get('pipeline_settings', {}) %}
 {%- for assigned_pipeline in ASSIGNED_PIPELINES %}
 - pipeline.id: {{ assigned_pipeline }}
   path.config: "/usr/share/logstash/pipelines/{{ assigned_pipeline }}/"
+{%-   set parsed = (PIPELINE_SETTINGS.get(assigned_pipeline, '') | load_yaml) or {} %}
+{#-   anything that is not a set of key/value pairs is ignored rather than allowed to
+      break this render; config.sls logs a warning for the same input #}
+{%-   set extra = parsed if parsed is mapping else {} %}
+{#-   pipeline.id and path.config are generated above; drop them so a pasted example
+      cannot repoint the pipeline or trigger a duplicate id error #}
+{%-   do extra.pop('pipeline.id', None) %}
+{%-   do extra.pop('path.config', None) %}
+{%-   if extra %}
+{{ extra | yaml(False) | indent(2, True) }}
+{%-   endif %}
 {% endfor -%}

salt/logstash/soc_logstash.yaml

@@ -16,6 +16,7 @@ logstash:
       heavynode: *assigned_pipelines
       searchnode: *assigned_pipelines
       manager: *assigned_pipelines
+      managerhype: *assigned_pipelines
       managersearch: *assigned_pipelines
       fleet: *assigned_pipelines
   defined_pipelines:
@@ -51,6 +52,32 @@ logstash:
     custom008: *pipeline_config
     custom009: *pipeline_config
     custom010: *pipeline_config
+  pipeline_settings:
+    manager: &pipeline_settings
+      description: >-
+        Pipeline-scoped Logstash settings for this pipeline, written in YAML, one setting
+        per line. For example, `pipeline.workers: 8`. These are added to this pipeline's
+        entry in pipelines.yml, and any setting left out here falls back to the value in
+        logstash.yml. Only pipeline-scoped settings are accepted; an unrecognized setting
+        name will stop every pipeline on this node from starting, and the error is written
+        to /opt/so/log/logstash/logstash.log rather than the container's console output.
+        pipeline.id and path.config are managed by Security Onion and are ignored if set
+        here.
+      advanced: True
+      global: False
+      multiline: True
+      syntax: yaml
+      forcedType: string
+      helpLink: logstash
+      duplicates: True
+    search: *pipeline_settings
+    receiver: *pipeline_settings
+    fleet: *pipeline_settings
+    custom0: *pipeline_settings
+    custom1: *pipeline_settings
+    custom2: *pipeline_settings
+    custom3: *pipeline_settings
+    custom4: *pipeline_settings
   settings:
     lsheap: 
       description: Heap size to use for logstash