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

VERSION

@@ -1 +1 @@
-3.0.0-foxtrot
+3.3.0

salt/elasticfleet/files/integrations/elastic-defend/elastic-defend-endpoints.json

@@ -5,7 +5,7 @@
   "package": {
     "name": "endpoint",
     "title": "Elastic Defend",
-    "version": "9.4.1",
+    "version": "9.3.1",
     "requires_root": true
   },
   "enabled": true,

salt/elasticfleet/files/integrations/grid-nodes_general/import-evtx-logs.json

@@ -29,7 +29,7 @@
               "\\.gz$"
             ],
             "include_files": [],
-            "processors": "- dissect:\n    tokenizer: \"/nsm/import/%{import.id}/evtx/%{import.file}\"\n    field: \"log.file.path\"\n    target_prefix: \"\"\n- decode_json_fields:\n    fields: [\"message\"]\n    target: \"\"\n- drop_fields:\n    fields: [\"host\"]\n    ignore_missing: true\n- add_fields:\n    target: data_stream\n    fields:\n      type: logs\n      dataset: system.security\n- add_fields:\n    target: event\n    fields:\n      dataset: system.security\n      module: system\n      imported: true\n- add_fields:\n    target: \"@metadata\"\n    fields:\n        pipeline: logs-system.security-2.22.3\n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-Sysmon/Operational'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: windows.sysmon_operational\n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.sysmon_operational\n          module: windows\n          imported: true\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-windows.sysmon_operational-3.9.0\n- if:\n    equals:\n      winlog.channel: 'Application'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: system.application\n    - add_fields:\n        target: event\n        fields:\n          dataset: system.application\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-system.application-2.22.3\n- if:\n    equals:\n      winlog.channel: 'System'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: system.system\n    - add_fields:\n        target: event\n        fields:\n          dataset: system.system\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-system.system-2.22.3\n    \n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-PowerShell/Operational'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: windows.powershell_operational\n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.powershell_operational\n          module: windows\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-windows.powershell_operational-3.9.0\n- add_fields:\n    target: data_stream\n    fields:\n        dataset: import",
+            "processors": "- dissect:\n    tokenizer: \"/nsm/import/%{import.id}/evtx/%{import.file}\"\n    field: \"log.file.path\"\n    target_prefix: \"\"\n- decode_json_fields:\n    fields: [\"message\"]\n    target: \"\"\n- drop_fields:\n    fields: [\"host\"]\n    ignore_missing: true\n- add_fields:\n    target: data_stream\n    fields:\n      type: logs\n      dataset: system.security\n- add_fields:\n    target: event\n    fields:\n      dataset: system.security\n      module: system\n      imported: true\n- add_fields:\n    target: \"@metadata\"\n    fields:\n        pipeline: logs-system.security-2.20.0\n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-Sysmon/Operational'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: windows.sysmon_operational\n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.sysmon_operational\n          module: windows\n          imported: true\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-windows.sysmon_operational-3.8.3\n- if:\n    equals:\n      winlog.channel: 'Application'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: system.application\n    - add_fields:\n        target: event\n        fields:\n          dataset: system.application\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-system.application-2.20.0\n- if:\n    equals:\n      winlog.channel: 'System'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: system.system\n    - add_fields:\n        target: event\n        fields:\n          dataset: system.system\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-system.system-2.20.0\n    \n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-PowerShell/Operational'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: windows.powershell_operational\n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.powershell_operational\n          module: windows\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-windows.powershell_operational-3.8.3\n- add_fields:\n    target: data_stream\n    fields:\n        dataset: import",
             "tags": [
               "import"
             ],

salt/elasticfleet/integration-defaults.map.jinja

@@ -16,6 +16,7 @@
     'awsfirehose.metrics': 'aws.cloudwatch',
     'cribl.logs': 'cribl',
     'cribl.metrics': 'cribl',
+    'sentinel_one_cloud_funnel.logins': 'sentinel_one_cloud_funnel.login',
     'azure_application_insights.app_insights': 'azure.app_insights',
     'azure_application_insights.app_state': 'azure.app_state',
     'azure_billing.billing': 'azure.billing',

salt/elasticsearch/cluster.sls

@@ -98,13 +98,6 @@ so-es-cluster-settings:
       - docker_container: so-elasticsearch
       - file: elasticsearch_sbin_jinja
       - http: wait_for_so-elasticsearch
-
-so-elasticsearch-system-indices-patch:
-  cmd.run:
-    - name: /usr/sbin/so-elasticsearch-system-indices-patch
-    - require:
-      - http: wait_for_so-elasticsearch
-      - file: so-elasticsearch-system-indices-patch-script
 {%  endif %}
 
 # heavynodes will only load ILM policies for SO managed indices. (Indicies defined in elasticsearch/defaults.yaml)

salt/elasticsearch/config.sls

@@ -42,16 +42,6 @@ elasticsearch_sbin:
     - file_mode: 755
     - exclude_pat:
       - so-elasticsearch-pipelines # exclude this because we need to watch it for changes, we sync it in another state
-      - so-elasticsearch-system-indices-patch
-    - show_changes: False
-
-so-elasticsearch-system-indices-patch-script:
-  file.managed:
-    - name: /usr/sbin/so-elasticsearch-system-indices-patch
-    - source: salt://elasticsearch/tools/sbin/so-elasticsearch-system-indices-patch
-    - user: 930
-    - group: 939
-    - mode: 755
     - show_changes: False
 
 elasticsearch_sbin_jinja:

salt/elasticsearch/defaults.yaml

@@ -1,7 +1,7 @@
 elasticsearch:
   enabled: false
   esheap: '600m'
-  version: 9.4.5
+  version: 9.3.7
   index_clean: true
   data_retention_method: DLM
   vm:

salt/elasticsearch/tools/sbin/so-elasticsearch-system-indices-patch

@@ -1,187 +0,0 @@
-#!/bin/bash
-# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
-# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
-# https://securityonion.net/license; you may not use this file except in compliance with the
-# Elastic License 2.0.
-
-set -eo pipefail
-
-SETTINGS='{"index":{"auto_expand_replicas":"0-1"}}'
-KIBANA_PASSWORD=
-INDEX_PATTERNS=(
-  '.entity_analytics.risk_score.lookup-*'
-  '.entity_analytics.watchlists.*'
-  '.workflows-executions'
-  '.workflows-step-executions'
-  '.entities.v2.latest.security_*'
-  '.entities.v2.history.security_*'
-  'risk-score.risk-score-latest-*'
-)
-DATA_STREAM_PATTERNS=(
-  '.entities.v2.updates.security_*'
-  'risk-score.risk-score-*'
-)
-TEMPLATE_PATTERNS=(
-  'entities_v2_latest_security_default_index_template'
-  'entities_v2_history_security_default_index_template'
-  '.entities_v2_updates_security_default_index_template'
-  '.risk-score.risk-score-default-index-template'
-)
-
-query_es() {
-  if so-elasticsearch-query "$@" --fail --retry 3 --retry-delay 5; then
-    return 0
-  fi
-
-  # retry failed attempts with so_kibana user (system managed indices reject so_elastic user)
-  local query_path="$1"
-  shift
-
-  if [[ -z "$KIBANA_PASSWORD" ]]; then
-    KIBANA_PASSWORD=$(salt-call pillar.get elasticsearch:auth:users:so_kibana_user:pass --out=newline_values_only)
-  fi
-  [[ -n "$KIBANA_PASSWORD" ]] || return 1
-
-  echo "Retrying ${query_path} as so_kibana." >&2
-  curl -K /opt/so/conf/elasticsearch/curl.config --user "so_kibana:${KIBANA_PASSWORD}" \
-    -s -k -L --fail --retry 3 --retry-delay 5 -H 'Content-Type: application/json' "https://localhost:9200/${query_path}" "$@"
-}
-
-# add auto_expand_replicas=0-1 to given index
-set_auto_expand_replicas() {
-  local index="$1"
-
-  echo "Setting auto_expand_replicas to 0-1 on ${index}."
-  query_es "${index}/_settings" -XPUT -d "$SETTINGS" >/dev/null
-}
-
-# resolve index patterns and find each index with an unassigned replica
-unassigned_replicas() {
-  local pattern="$1"
-  local resolved_indices response index
-
-  if ! resolved_indices=$(query_es "_resolve/index/${pattern}?expand_wildcards=all" 2>/dev/null); then
-    return 0
-  fi
-
-  while read -r index; do
-    if ! response=$(query_es "_cat/shards/${index}?format=json&h=index,prirep,state" 2>/dev/null); then
-      continue
-    fi
-    jq -r '.[]? | objects | select(.prirep == "r" and .state == "UNASSIGNED") | .index' <<<"$response"
-  done < <(jq -r '.indices[]?.name' <<<"$resolved_indices")
-}
-
-data_stream_indices() {
-  local pattern="$1"
-  local response
-
-  if ! response=$(query_es "_data_stream/${pattern}?expand_wildcards=all" 2>/dev/null); then
-    return 0
-  fi
-  jq -r '.data_streams[]?.indices[]?.index_name' <<<"$response"
-}
-
-update_system_indices() {
-  local pattern="$1"
-  local index
-
-  while read -r index; do
-    [[ -n "$index" ]] && set_auto_expand_replicas "$index"
-  done < <(unassigned_replicas "$pattern")
-}
-
-# update data stream backing indices with unassigned replicas
-update_system_ds() {
-  local pattern="$1"
-  local index
-
-  while read -r index; do
-    while read -r unassigned_index; do
-      [[ -n "$unassigned_index" ]] && set_auto_expand_replicas "$unassigned_index"
-    done < <(unassigned_replicas "$index")
-  done < <(data_stream_indices "$pattern")
-}
-
-has_unassigned_replicas() {
-  local pattern="$1"
-  local index
-
-  index=$(unassigned_replicas "$pattern" | sed -n '1p')
-  [[ -n "$index" ]]
-}
-
-data_stream_has_unassigned_replicas() {
-  local pattern="$1"
-  local index
-  while read -r index; do
-    has_unassigned_replicas "$index" && return 0
-  done < <(data_stream_indices "$pattern")
-
-  return 1
-}
-
-needs_patch() {
-  local pattern
-  for pattern in "${INDEX_PATTERNS[@]}"; do
-    has_unassigned_replicas "$pattern" && return 0
-  done
-  for pattern in "${DATA_STREAM_PATTERNS[@]}"; do
-    data_stream_has_unassigned_replicas "$pattern" && return 0
-  done
-
-  return 1
-}
-
-# get index templates, update with auto_expand_replicas=0-1, and PUT back. Keeping mappings/settings/aliases in-place
-update_system_templates() {
-  local pattern="$1"
-  local templates name response template auto_expand_replicas
-
-  if ! templates=$(query_es "_index_template/${pattern}" 2>/dev/null); then
-    return 0
-  fi
-  while read -r name; do
-    response=$(query_es "_index_template/${name}")
-    template=$(jq -c '.index_templates[0].index_template' <<<"$response")
-    auto_expand_replicas=$(jq -r '.template.settings["index.auto_expand_replicas"] // .template.settings.index.auto_expand_replicas // empty' <<<"$template")
-    [[ "$auto_expand_replicas" == "0-1" ]] && continue
-
-    template=$(jq '
-            if (.template.settings.index | type) == "object" then
-                .template.settings.index.auto_expand_replicas = "0-1"
-            else
-                .template.settings["index.auto_expand_replicas"] = "0-1"
-            end
-            | del(.created_date_millis, .modified_date_millis)
-        ' <<<"$template")
-    echo "Setting auto_expand_replicas to 0-1 on index template ${name}."
-    query_es "_index_template/${name}" -XPUT -d "$template" >/dev/null
-  done < <(jq -r '.index_templates[]?.name' <<<"$templates")
-}
-
-if [[ "${1:-}" == "--check" ]]; then
-  needs_patch
-  exit $?
-fi
-
-if [[ $# -ne 0 ]]; then
-  echo "Usage: $0 [--check]" >&2
-  exit 1
-fi
-
-for pattern in "${INDEX_PATTERNS[@]}"; do
-  if has_unassigned_replicas "$pattern"; then
-    update_system_indices "$pattern"
-  fi
-done
-
-for pattern in "${DATA_STREAM_PATTERNS[@]}"; do
-  if data_stream_has_unassigned_replicas "$pattern"; then
-    update_system_ds "$pattern"
-  fi
-done
-
-for pattern in "${TEMPLATE_PATTERNS[@]}"; do
-  update_system_templates "$pattern"
-done

salt/kibana/defaults.yaml

@@ -22,7 +22,7 @@ kibana:
           - default
           - file
     migrations:
-      discardCorruptObjects: "9.4.5"
+      discardCorruptObjects: "9.3.7"
     telemetry:
       enabled: False
     xpack:

salt/kibana/tools/sbin_jinja/so-kibana-space-defaults

@@ -9,5 +9,5 @@ SESSIONCOOKIE=$(curl -K /opt/so/conf/elasticsearch/curl.config -c - -X GET http:
 # Disable certain Features from showing up in the Kibana UI
 echo
 echo "Setting up default Kibana Space:"
-curl -K /opt/so/conf/elasticsearch/curl.config -b "sid=$SESSIONCOOKIE" -L -X PUT "localhost:5601/api/spaces/space/default" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d' {"id":"default","name":"Default","disabledFeatures":["ml","enterpriseSearch","logs","infrastructure","apm","uptime","securitySolutionCasesV3","inventory","searchSynonyms","searchQueryRules","enterpriseSearchApplications","enterpriseSearchAnalytics","securitySolutionTimeline","securitySolutionNotes","securitySolutionRulesV4","securitySolutionAlertsV1","entityManager","slo","streams","anonymization","searchInferenceEndpoints","cloudConnect","queryActivity","automatic_import","stackAlerts","monitoring","dataQuality","actions"]} ' >> /opt/so/log/kibana/misc.log
+curl -K /opt/so/conf/elasticsearch/curl.config -b "sid=$SESSIONCOOKIE" -L -X PUT "localhost:5601/api/spaces/space/default" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d' {"id":"default","name":"Default","disabledFeatures":["ml","enterpriseSearch","logs","infrastructure","apm","uptime","monitoring","stackAlerts","actions","securitySolutionCasesV3","inventory","dataQuality","searchSynonyms","searchQueryRules","enterpriseSearchApplications","enterpriseSearchAnalytics","securitySolutionTimeline","securitySolutionNotes","securitySolutionRulesV1","entityManager","streams","cloudConnect","slo"]} ' >> /opt/so/log/kibana/misc.log
 echo

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

salt/soc/defaults.yaml

@@ -2671,7 +2671,7 @@ soc:
               # The id (UUIDv4) is pregenerated and can safely be used.
               # Click "Convert" to convert the Sigma rule to use Security Onion field mappings within an EQL query
               #
-              # Rule Creation Guide: https://github.com/SigmaHQ/sigma/wiki/Rule-Creation-High%E2%80%90Level-Guide
+              # Rule Creation Guide: https://github.com/SigmaHQ/sigma/wiki/Rule-Creation-Guide
               # Logsources: https://sigmahq.io/docs/basics/log-sources.html
 
               title: 'A Short Capitalized Title With Less Than 50 Characters'
@@ -2683,7 +2683,7 @@ soc:
               references:
                 - 'https://local.invalid'
               author: '@SecurityOnion'
-              date: '[today]'
+              date: 'YYYY/MM/DD'
               tags:
                 - detection.threat_hunting
                 - attack.technique_id