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

salt/salt/minion/boot_highstate.sls

@@ -3,6 +3,8 @@
 # https://securityonion.net/license; you may not use this file except in compliance with the
 # Elastic License 2.0.
 
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+
 # Manages /etc/systemd/system/so-boot-highstate.service, a Type=oneshot
 # RemainAfterExit=yes unit that runs `salt-call state.highstate` exactly once
 # per system boot. Replaces the legacy `startup_states: highstate` minion
@@ -19,9 +21,25 @@ so_boot_highstate_unit_file:
     - onchanges_in:
       - module: systemd_reload
 
+# Non-managers never apply salt.minion during setup, so reaching this state means
+# setup is finished and the marker is safe to write unconditionally. This also
+# heals nodes installed before this fix, which have no marker and no legacy
+# startup_states line to grep for. Managers do highstate mid-setup, so they only
+# get the marker from the legacy upgrade signal; fresh installs get it from
+# mark_setup_complete in setup/so-functions.
+mark_setup_complete:
+  file.managed:
+    - name: /opt/so/state/setup-complete
+    - replace: false
+    - makedirs: True
+{% if GLOBALS.is_manager %}
+    - onlyif: "grep -qx 'startup_states: highstate' /etc/salt/minion"
+{% endif %}
+    - require_in:
+      - service: so_boot_highstate_service
+
 # Only enable once setup is complete. Until then the gate file is missing and
-# the unit's own ConditionPathExists would no-op it anyway -- this just keeps
-# `systemctl is-enabled` honest for the sync_es_users gate.
+# the unit's own ConditionPathExists would no-op it anyway.
 so_boot_highstate_service:
   service.enabled:
     - name: so-boot-highstate.service

salt/salt/minion/init.sls

@@ -87,27 +87,15 @@ set_log_levels:
 # so-boot-highstate.service (managed in salt.minion.boot_highstate), which
 # runs once per system boot only. Strip the line from /etc/salt/minion on
 # upgrade; both the commented and uncommented forms historically existed.
+# Ordered after mark_setup_complete (salt.minion.boot_highstate); the manager
+# gate there greps for this line, so it must run before we delete it.
 remove_startup_states:
   file.line:
     - name: /etc/salt/minion
     - match: 'startup_states: highstate'
     - mode: delete
-
-# Upgrade-path bridge: systems that already passed setup under the old gate
-# (`grep -x 'startup_states: highstate' /etc/salt/minion`) get a /opt/so/state/setup-complete
-# marker so so-boot-highstate.service can be enabled and the so-user_sync cron
-# in sync_es_users.sls keeps installing. Setup-in-progress systems instead get
-# the marker from `mark_setup_complete` in setup/so-functions at the right
-# moment. `replace: false` means we never overwrite a marker once written.
-mark_setup_complete_for_upgrades:
-  file.managed:
-    - name: /opt/so/state/setup-complete
-    - replace: false
-    - makedirs: True
-    - onlyif: "grep -qx 'startup_states: highstate' /etc/salt/minion"
-    - require_in:
-      - file: remove_startup_states
-      - service: so_boot_highstate_service
+    - require:
+      - file: mark_setup_complete
 
 {% endif %}
 

salt/setup/virt/setSalt.sls

@@ -8,6 +8,15 @@ set_role_grain:
     - name: role
     - value: so-{{ grains.id.split("_") | last }}
 
+# salt-cloud guests never run so-setup, so nothing else marks them setup-complete.
+# Replaces the 'startup_states: highstate' line this state used to append. No
+# GLOBALS import -- this runs before the guest's pillars exist.
+mark_setup_complete_vm_guest:
+  file.managed:
+    - name: /opt/so/state/setup-complete
+    - replace: false
+    - makedirs: True
+
 enable_salt_minion:
   service.enabled:
     - name: salt-minion

salt/soc/defaults.yaml

@@ -1537,18 +1537,6 @@ soc:
             Orchestrator: sonnet@SOAI
             Investigator: gemma@SOAI
             DetectionEngineer: gemma@SOAI
-          useMemory: true
-          useMemoryScanner: false
-          memoryScanIntervalSeconds: 300
-          memoryProximityThreshold: 0.8
-          messageProximityThreshold: 0.5
-          maxUserMemoriesToInclude: 5
-          maxGlobalMemoriesToInclude: 5
-          maxUserMemoriesToReconcile: 20
-          maxGlobalMemoriesToReconcile: 20
-          memoryModel: gemma@SOAI
-          embedModel: amazon.titan-embed-text-v2@SOAI
-          reconcileModel: gemma@SOAI
         onionconfig:
           saltstackDir: /opt/so/saltstack
           bypassEnabled: false
@@ -2739,14 +2727,5 @@ soc:
               enabled: true
               adapter: SOAI
               charsPerTokenEstimate: 4
-            - id: amazon.titan-embed-text-v2
-              displayName: amazon.titan-embed-text-v2
-              origin: USA
-              contextLimitSmall: 8192
-              contextLimitLarge: 8192
-              lowBalanceColorAlert: 500000
-              enabled: true
-              adapter: SOAI
-              charsPerTokenEstimate: 4
               
             

salt/soc/soc_soc.yaml

@@ -845,42 +845,6 @@ soc:
             DetectionEngineer:
               description: This agent manages detections and their overrides, including tuning noisy rules and authoring rule content.
               global: True
-          useMemory:
-            description: Enables the Memory system for OnionAI
-            global: True
-          useMemoryScanner:
-            description: Enables the memory scanner for automatic memory extraction from historical sessions.
-            global: True
-          memoryScanIntervalSeconds:
-            description: How long to wait in seconds between attempts to scan sessions for new memories.
-            global: True
-          memoryProximityThreshold:
-            description: Describes how close memories need to be on a floating point scale from 0.0 to 1.0 to be considered when reconciling new memories with old ones. This value is usually higher than messageProximityThreshold.
-            global: True
-          messageProximityThreshold:
-            description: Describes how close a memory needs to be to a user's message on a floating point scale from 0.0 to 1.0 to be included in the context. This value is usually lower than memoryProximityThreshold.
-            global: True
-          maxUserMemoriesToInclude:
-            description: Specify the max number of user-specific memories to include in the prompt when a user sends a message.
-            global: True
-          maxGlobalMemoriesToInclude:
-            description: Specify the max number of global memories to include in the prompt when a user sends a message.
-            global: True
-          maxUserMemoriesToReconcile:
-            description: When reconciling new user-specific memories with existing user-specific memories, this determines how many old memories may be considered.
-            global: True
-          maxGlobalMemoriesToReconcile:
-            description: When reconciling new global memories with existing global memories, this determines how many old memories may be considered.
-            global: True
-          memoryModel:
-            description: The model to use when extracting memories from sessions.
-            global: True
-          embedModel: 
-            description: The model to use when embedding a memory as a vector. Note that only memories embedded using the same model may be compared and only memories created with the model specified here will be considered when informing an agent of existing memories.
-            global: True
-          reconcileModel:
-            description: The model to use when reconciling memories that contain nearly the same content.
-            global: True
       client:
         assistant:
           enabled:

setup/so-setup

@@ -833,6 +833,7 @@ if ! [[ -f $install_opt_file ]]; then
 		check_sos_appliance
 		drop_install_options
 		hypervisor_local_states
+		mark_setup_complete
 		verify_setup
 	fi