From bf249b7adf6647cfc3028bf675fe2bb31066a68e Mon Sep 17 00:00:00 2001 From: Josh Patterson Date: Wed, 19 Aug 2026 15:33:26 -0400 Subject: [PATCH] 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 | 12 ++++++++++-- salt/logstash/defaults.yaml | 10 ++++++++++ salt/logstash/enabled.sls | 1 + salt/logstash/etc/pipelines.yml.jinja | 13 +++++++++++++ salt/logstash/soc_logstash.yaml | 27 +++++++++++++++++++++++++++ 5 files changed, 61 insertions(+), 2 deletions(-) diff --git a/salt/logstash/config.sls b/salt/logstash/config.sls index 47feba42c..79c9e01d0 100644 --- a/salt/logstash/config.sls +++ b/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 %} diff --git a/salt/logstash/defaults.yaml b/salt/logstash/defaults.yaml index db5e4ee58..a76f5a804 100644 --- a/salt/logstash/defaults.yaml +++ b/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: diff --git a/salt/logstash/enabled.sls b/salt/logstash/enabled.sls index 80e40d78d..fb0852f00 100644 --- a/salt/logstash/enabled.sls +++ b/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 diff --git a/salt/logstash/etc/pipelines.yml.jinja b/salt/logstash/etc/pipelines.yml.jinja index 427cc9f14..4fa6114ae 100644 --- a/salt/logstash/etc/pipelines.yml.jinja +++ b/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 -%} diff --git a/salt/logstash/soc_logstash.yaml b/salt/logstash/soc_logstash.yaml index 40794afe4..b85ea563c 100644 --- a/salt/logstash/soc_logstash.yaml +++ b/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