Stop unused Fleet latest transforms that hold empty shards

After addon package install, stop transforms for packages that are not
on an enabled agent policy. On soup/highstate, stop remaining latest
transforms whose source and destination are empty and whose package is
not assigned to a policy. An operator start in Kibana is left running.

.github/DISCUSSION_TEMPLATE/3-0.yml

@@ -14,7 +14,6 @@ body:
         - 3.2.0
         - 3.3.0
         - 3.4.0
-        - 3.5.0
         - Other (please provide detail below)
     validations:
       required: true

DOWNLOAD_AND_VERIFY_ISO.md

@@ -1,17 +1,17 @@
-### 3.4.0-20261102 ISO image released on 2026/11/02
+### 3.3.0-20260911 ISO image released on 2026/09/11
 
 
 ### Download and Verify
 
-3.4.0-20261102 ISO image:  
+3.3.0-20260911 ISO image:  
-https://download.securityonion.net/file/securityonion/securityonion-3.4.0-20261102.iso
+https://download.securityonion.net/file/securityonion/securityonion-3.3.0-20260911.iso
  
-MD5: FB0380AE996E175D6CC7E7A35D8CC178  
+MD5: 12B18433D3A2198A185892FF79CF638F  
-SHA1: D8960F1DA352710BCA209773BF0B3AE8791D0D1B  
+SHA1: 2B3C2E1FA7A78ED1F956E7EDCC12E32593C14EEE  
-SHA256: DFC4C87668836852A328EC7A0B755C8C1511FF0E43D423DEE23047A95F676D99  
+SHA256: 0938C73B76CE30EC9E4394D312C79EA7CAC721B6818541697279A6221F7D870D  
 
 Signature for ISO image:  
-https://github.com/Security-Onion-Solutions/securityonion/raw/3/main/sigs/securityonion-3.4.0-20261102.iso.sig
+https://github.com/Security-Onion-Solutions/securityonion/raw/3/main/sigs/securityonion-3.3.0-20260911.iso.sig
 
 Signing key:  
 https://raw.githubusercontent.com/Security-Onion-Solutions/securityonion/3/main/KEYS  
@@ -25,22 +25,22 @@ wget https://raw.githubusercontent.com/Security-Onion-Solutions/securityonion/3/
 
 Download the signature file for the ISO:  
 ```
-wget https://github.com/Security-Onion-Solutions/securityonion/raw/3/main/sigs/securityonion-3.4.0-20261102.iso.sig
+wget https://github.com/Security-Onion-Solutions/securityonion/raw/3/main/sigs/securityonion-3.3.0-20260911.iso.sig
 ```
 
 Download the ISO image:  
 ```
-wget https://download.securityonion.net/file/securityonion/securityonion-3.4.0-20261102.iso
+wget https://download.securityonion.net/file/securityonion/securityonion-3.3.0-20260911.iso
 ```
 
 Verify the downloaded ISO image using the signature file:  
 ```
-gpg --verify securityonion-3.4.0-20261102.iso.sig securityonion-3.4.0-20261102.iso
+gpg --verify securityonion-3.3.0-20260911.iso.sig securityonion-3.3.0-20260911.iso
 ```
 
 The output should show "Good signature" and the Primary key fingerprint should match what's shown below:
 ```
-gpg: Signature made Sat 31 Oct 2026 09:27:50 AM EDT using RSA key ID FE507013
+gpg: Signature made Fri 11 Sep 2026 11:23:56 AM EDT using RSA key ID FE507013
 gpg: Good signature from "Security Onion Solutions, LLC <info@securityonionsolutions.com>"
 gpg: WARNING: This key is not certified with a trusted signature!
 gpg:          There is no indication that the signature belongs to the owner.

VERSION

@@ -1 +1 @@
-3.5.0
+3.4.0

salt/_beacons/postgres_pillar_beacon.py

@@ -131,8 +131,6 @@ def beacon(config):  # noqa: C901
             'setting_id': setting_id,
             'node_id': node_id,
         })
-        log.info('postgres_pillar_beacon: audit_settings id=%d setting_id=%s node_id=%s',
-                 row_id, setting_id, node_id)
         if row_id > max_id:
             max_id = row_id
 

salt/desktop/packages.sls

@@ -438,10 +438,6 @@ install_networkminer:
   pkg.latest:
     - name: securityonion-networkminer
 
-install_so_crates:
-  pkg.latest:
-    - name: securityonion-so-crates
-
 {% else %}
 
 desktop_packages_os_fail:

salt/elasticfleet/manager.sls

@@ -112,6 +112,17 @@ so-elastic-fleet-addon-integrations:
       - http: wait_for_so-kibana
       - cmd: so-elasticsearch-templates-after-fleet-packages
 
+so-elastic-fleet-transforms-manage:
+  cmd.run:
+    - name: /usr/sbin/so-elastic-fleet-transforms-manage
+    - require:
+      - http: wait_for_so-kibana
+      - file: elasticfleet_sbin
+      - cmd: so-elastic-fleet-package-upgrade
+      - cmd: so-elastic-fleet-integration-upgrade
+      - cmd: so-elastic-fleet-integrations
+      - cmd: so-elastic-fleet-addon-integrations
+
 {% if ELASTICFLEETMERGED.config.defend_filters.enable_auto_configuration %}
 so-elastic-defend-manage-filters-file-watch:
   cmd.run:

salt/elasticfleet/tools/sbin/so-elastic-fleet-transforms-manage

@@ -0,0 +1,225 @@
+#!/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.
+#
+# 1. List transforms (source/dest come from the transform config).
+# 2. List enabled Fleet package policies.
+# 3. Unused packages: count only those source/dest indices, then stop and
+#    delete empty dests. --after-install skips the counts and stops unused.
+
+. /usr/sbin/so-common
+. /usr/sbin/so-elastic-fleet-common
+
+PIVOT_PREFIX='logs-elasticsearch.index_pivot-default-'
+STATE_FILE=/opt/so/state/esfleet_stopped_transforms.json
+DRY_RUN=false
+AFTER_INSTALL=false
+
+usage() {
+    echo "Usage: $0 [--dry-run] [--after-install]" >&2
+    exit 1
+}
+
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        --dry-run) DRY_RUN=true ;;
+        --after-install) AFTER_INSTALL=true ;;
+        -h|--help) usage ;;
+        *) usage ;;
+    esac
+    shift
+done
+
+es() {
+    so-elasticsearch-query "$@" --fail --retry 2 --retry-delay 3
+}
+
+skip() {
+    echo "Warning: $1; skipping transform cleanup." >&2
+    exit 0
+}
+
+run() {
+    local desc=$1
+    shift
+    if [[ "$DRY_RUN" == true ]]; then
+        echo "[dry-run] $desc"
+        return 0
+    fi
+    echo "$desc"
+    es "$@" >/dev/null
+}
+
+uri() {
+    jq -rn --arg id "$1" '$id|@uri'
+}
+
+in_use_packages() {
+    local page=1 collected=0 count total resp packages='[]'
+    while true; do
+        resp=$(fleet_api "package_policies?perPage=100&page=$page") || return 1
+        jq -e '(.items | type == "array") and (.total | type == "number")' <<<"$resp" >/dev/null || return 1
+        packages=$(jq -c --argjson acc "$packages" '
+            $acc + [.items[]
+              | select(.enabled == true)
+              | select(([.policy_ids[]?, .policy_id] | map(select(. != null and . != "")) | length) > 0)
+              | .package.name
+              | select(. != null and . != "")]' <<<"$resp")
+        count=$(jq '.items | length' <<<"$resp")
+        total=$(jq '.total' <<<"$resp")
+        collected=$((collected + count))
+        [[ "$count" -eq 0 || "$collected" -ge "$total" ]] && break
+        page=$((page + 1))
+        [[ "$page" -gt 100 ]] && return 1
+    done
+    [[ "$collected" -eq "$total" ]] || return 1
+    jq -c 'unique' <<<"$packages"
+}
+
+# Page into files: a full transform list exceeds ARG_MAX.
+page_es() {
+    local path=$1 outfile=$2 from=0 size=1000 count got page_len
+    echo '[]' > "$outfile"
+    while true; do
+        es "${path}?from=${from}&size=${size}" > "$WORKDIR/page.json" || return 1
+        jq -e '(.count | type == "number") and (.transforms | type == "array")' "$WORKDIR/page.json" >/dev/null || return 1
+        count=$(jq -r '.count' "$WORKDIR/page.json")
+        jq -c --slurpfile page "$WORKDIR/page.json" '. + ($page[0].transforms // [])' "$outfile" > "$WORKDIR/next.json" || return 1
+        mv "$WORKDIR/next.json" "$outfile"
+        got=$(jq 'length' "$outfile")
+        page_len=$(jq '.transforms | length' "$WORKDIR/page.json")
+        [[ "$page_len" -eq 0 || "$got" -ge "$count" ]] && break
+        from=$((from + size))
+        [[ "$from" -gt 100000 ]] && return 1
+    done
+    [[ "$got" -eq "$count" ]]
+}
+
+load_transforms() {
+    page_es "_transform" "$WORKDIR/tf-configs.json" || return 1
+    page_es "_transform/_stats" "$WORKDIR/tf-stats.json" || return 1
+    jq -c --slurpfile stats "$WORKDIR/tf-stats.json" --arg pivot "$PIVOT_PREFIX" '
+        ($stats[0] | map({key: .id, value: .state}) | from_entries) as $states
+        | [.[] | {
+            id,
+            package: (._meta.package.name // null),
+            state: ($states[.id] // "unknown"),
+            dest: (.dest.index // null),
+            create_time: (.create_time // null),
+            source: (.source.index | if type == "array" then . elif type == "string" then [.] else [] end),
+            pivot: (.id | startswith($pivot)),
+            latest: ((.latest != null) or ((.dest.index // "") | contains("_latest")))
+          }
+          | select(.pivot or .latest)
+        ]' "$WORKDIR/tf-configs.json" > "$WORKDIR/transforms.json"
+}
+
+doc_count() {
+    local pattern=$1 count
+    count=$(es "${pattern}/_count?ignore_unavailable=true" | jq -er '.count') || return 1
+    [[ "$count" =~ ^[0-9]+$ ]] || return 1
+    echo "$count"
+}
+
+status "Elastic Fleet transform cleanup (after_install=$AFTER_INSTALL dry_run=$DRY_RUN)"
+
+WORKDIR=$(mktemp -d)
+trap 'rm -rf "$WORKDIR"' EXIT
+
+load_transforms || skip "Elasticsearch transforms could not be retrieved in full"
+IN_USE=$(in_use_packages) || skip "Fleet package policies could not be retrieved in full"
+jq -e 'length > 0' <<<"$IN_USE" >/dev/null || skip "no enabled Fleet package policies are assigned to an agent policy"
+
+STATE=$(jq -c '{stopped: ((.stopped | objects) // {}), user_started: ((.user_started | arrays) // [])}' "$STATE_FILE" 2>/dev/null)
+jq -e '.stopped and .user_started' <<<"$STATE" >/dev/null 2>&1 || STATE='{"stopped":{},"user_started":[]}'
+
+# Same create_time but running again: operator started it after we stopped it.
+STATE=$(jq -c --slurpfile t "$WORKDIR/transforms.json" '
+    ($t[0] | map({key: .id, value: .}) | from_entries) as $cur
+    | (.stopped | to_entries | map(select($cur[.key] != null and $cur[.key].create_time == .value))) as $tracked
+    | {
+        stopped: ($tracked | map(select($cur[.key].state == "stopped")) | from_entries),
+        user_started: ((.user_started + [$tracked[] | select($cur[.key].state | IN("started", "indexing")) | .key])
+                       | unique | map(select($cur[.] != null and ($cur[.].state | IN("stopped", "failed") | not))))
+      }' <<<"$STATE")
+
+# Unused = latest/pivot with a package, not on a policy, not operator-started.
+# Pivot is always unused (see so-elastic-fleet-package-upgrade).
+jq -c --argjson inuse "$IN_USE" --argjson state "$STATE" '
+    def unused:
+      if .pivot then true
+      elif .package == null then false
+      elif (.id | IN($state.user_started[]?)) then false
+      elif (.package | IN($inuse[]?)) then false
+      else true
+      end;
+    {
+      unused: [.[] | select(unused)],
+      start: [.[] | select(.package != null and (.state | IN("stopped", "failed"))
+                           and ($state.stopped[.id] != null) and (.pivot | not)
+                           and (.package | IN($inuse[]?)))]
+    }' "$WORKDIR/transforms.json" > "$WORKDIR/split.json"
+
+jq -c '.unused' "$WORKDIR/split.json" > "$WORKDIR/unused.json"
+
+if [[ "$AFTER_INSTALL" == true ]]; then
+    jq -c '[.[] | select(.state | IN("started", "indexing", "failed"))]' "$WORKDIR/unused.json" > "$WORKDIR/stop.json"
+else
+    echo '{}' > "$WORKDIR/counts.json"
+    while IFS= read -r pattern; do
+        [[ -n "$pattern" ]] || continue
+        n=$(doc_count "$pattern") || skip "could not count documents in $pattern"
+        jq -c --arg p "$pattern" --argjson n "$n" '. + {($p): $n}' "$WORKDIR/counts.json" > "$WORKDIR/counts-next.json"
+        mv "$WORKDIR/counts-next.json" "$WORKDIR/counts.json"
+    done < <(jq -r '[.[].dest | select(. != null and . != "")] + [.[].source[]?] | unique | .[]' "$WORKDIR/unused.json")
+
+    jq -c --slurpfile c "$WORKDIR/counts.json" '
+        ($c[0] // {}) as $n
+        | [.[] | select(.state | IN("started", "indexing", "failed"))
+             | select((if .dest == null or .dest == "" then 0 else ($n[.dest] // 0) end) == 0)
+             | select(([.source[]? | $n[.] // 0] | add // 0) == 0)]' \
+        "$WORKDIR/unused.json" > "$WORKDIR/stop.json"
+fi
+
+echo "in_use=$(jq -r 'join(",")' <<<"$IN_USE") unused=$(jq 'length' "$WORKDIR/unused.json") stop=$(jq 'length' "$WORKDIR/stop.json") start=$(jq '.start | length' "$WORKDIR/split.json") dry_run=$DRY_RUN"
+
+while IFS= read -r id; do
+    [[ -n "$id" ]] || continue
+    force=
+    jq -e --arg id "$id" '.[] | select(.id == $id and .state == "failed")' "$WORKDIR/stop.json" >/dev/null && force='&force=true'
+    if run "stop transform $id" "_transform/$(uri "$id")/_stop?wait_for_completion=true&timeout=60s${force}" -XPOST; then
+        [[ "$DRY_RUN" == true ]] && continue
+        ct=$(jq -c --arg id "$id" '.[] | select(.id == $id) | .create_time' "$WORKDIR/stop.json")
+        STATE=$(jq -c --arg id "$id" --argjson ct "$ct" '.stopped[$id] = $ct | .user_started -= [$id]' <<<"$STATE")
+    fi
+done < <(jq -r '.[].id' "$WORKDIR/stop.json")
+
+while IFS= read -r id; do
+    [[ -n "$id" ]] || continue
+    if run "start transform $id" "_transform/$(uri "$id")/_start" -XPOST; then
+        [[ "$DRY_RUN" == false ]] && STATE=$(jq -c --arg id "$id" 'del(.stopped[$id])' <<<"$STATE")
+    fi
+done < <(jq -r '.start[].id' "$WORKDIR/split.json")
+
+if [[ "$DRY_RUN" == false ]]; then
+    mkdir -p "$(dirname "$STATE_FILE")"
+    tmp=$(mktemp "${STATE_FILE}.XXXXXX") && printf '%s\n' "$STATE" > "$tmp" && mv -f "$tmp" "$STATE_FILE"
+fi
+
+DELETED=0
+while IFS= read -r dest; do
+    [[ -n "$dest" ]] || continue
+    if [[ "$DRY_RUN" == true ]]; then
+        echo "[dry-run] delete empty transform destination $dest"
+        DELETED=$((DELETED + 1))
+        continue
+    fi
+    n=$(doc_count "$dest") || continue
+    [[ "$n" -eq 0 ]] || continue
+    run "delete empty transform destination $dest" "$(uri "$dest")" -XDELETE && DELETED=$((DELETED + 1))
+done < <(jq -r '.[].dest | select(. != null and . != "")' "$WORKDIR/stop.json" | sort -u)
+
+echo "stopped=$(jq -r '.stopped | length' <<<"$STATE") user_started=$(jq -r '.user_started | length' <<<"$STATE") dests_removed=$DELETED"

salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-optional-integrations-load

@@ -137,6 +137,7 @@ def needs($sub;$autoup;$excluded;$inuse):
             done
             # cleanup any temp files for chunked package install
             rm -f ${pkg_filename}_*.json $BULK_INSTALL_PACKAGE_LIST
+            /usr/sbin/so-elastic-fleet-transforms-manage --after-install || echo "Warning: transform cleanup after package install failed; highstate will retry."
         else
             echo "Elastic integrations don't appear to need installation/updating..."
         fi

salt/elasticsearch/cluster.sls

@@ -107,12 +107,6 @@ so-elasticsearch-system-indices-patch:
       - file: so-elasticsearch-system-indices-patch-script
 {%  endif %}
 
-so-elasticsearch-ilm-policy-reload:
-  file.absent:
-    - name: /opt/so/state/esilm.txt
-    - onchanges:
-      - file: so-elasticsearch-ilm-policy-load-script
-
 # heavynodes will only load ILM policies for SO managed indices. (Indicies defined in elasticsearch/defaults.yaml)
 so-elasticsearch-ilm-policy-load:
   cmd.run:
@@ -121,9 +115,8 @@ so-elasticsearch-ilm-policy-load:
     - require:
       - docker_container: so-elasticsearch
       - file: so-elasticsearch-ilm-policy-load-script
-      - http: wait_for_so-elasticsearch
+    - onchanges:
-      - file: so-elasticsearch-ilm-policy-reload
+      - file: so-elasticsearch-ilm-policy-load-script
-    - unless: test -f /opt/so/state/esilm.txt
 
 so-elasticsearch-templates-reload:
   file.absent:

salt/elasticsearch/tools/sbin_jinja/so-elasticsearch-ilm-policy-load

@@ -7,17 +7,11 @@
 . /usr/sbin/so-common
 
 MAX_JOBS=${MAX_ILM_JOBS:-10}
-ILM_STATEFILE=/opt/so/state/esilm.txt
-rm -f "$ILM_STATEFILE"
 
-check_elasticsearch_responsive
-
-ILM_TMP_DIR=$(mktemp -d)
-trap 'rm -rf "$ILM_TMP_DIR"' EXIT
 # Lock used to serialize block writes so concurrent jobs never interleave their output.
-ILM_OUTPUT_LOCK="$ILM_TMP_DIR/output.lock"
+ILM_OUTPUT_LOCK=$(mktemp)
-ILM_FAIL_FILE="$ILM_TMP_DIR/failed"
+ILM_FAIL_FILE=$(mktemp)
-touch "$ILM_FAIL_FILE"
+trap 'rm -f "$ILM_OUTPUT_LOCK" "$ILM_FAIL_FILE"' EXIT
 
 # Policies are loaded concurrently (up to MAX_JOBS at a time) for speed. Each policy's block is
 # printed the moment its curl returns, so output appears in COMPLETION ORDER, not the order
@@ -26,30 +20,13 @@ echo "Loading ILM policies concurrently; output below appears in completion orde
 echo
 
 put_policy() {
-  local desc="$1" policyname="$2" data="$3" rc=1 attempt out http_code body result reason tmpdata
+  local desc="$1" policyname="$2" data="$3" result rc=0
-  tmpdata=$(mktemp "$ILM_TMP_DIR/${policyname}.XXXXXX.json")
+  if ! result=$(curl -K /opt/so/conf/elasticsearch/curl.config -s -k -L --fail \
-  printf '%s' "$data" >"$tmpdata"
+    -X PUT "https://localhost:9200/_ilm/policy/${policyname}" \
-
+    -H 'Content-Type: application/json' -d"${data}" 2>&1); then
-  for attempt in 1 2 3; do
+    rc=1
-    out=$(so-elasticsearch-query "_ilm/policy/${policyname}" -d@"${tmpdata}" -XPUT -w '\n%{http_code}' 2>/dev/null)
+  elif ! jq -e '.acknowledged == true' <<<"$result" >/dev/null 2>&1; then
-    http_code=${out##*$'\n'} body=${out%$'\n'*}
+    rc=1
-
-    if [[ "$http_code" == 200 ]] && jq -e '.acknowledged == true' <<<"$body" >/dev/null 2>&1; then
-      rc=0
-      break
-    fi
-    # 429 = ES rejected/throttled (may succeed later); other 4xx will not
-    [[ "$http_code" == 4?? && "$http_code" != 429 ]] && break
-    (( attempt < 3 )) && sleep 3
-  done
-
-  if (( rc == 0 )); then
-    result='{"acknowledged":true}'
-  else
-    result=${body:-"HTTP ${http_code:-000}"}
-    # flatten tabs/newlines so the TAB fail-file stays one record
-    reason=$(jq -r '(if (.error|type) == "object" then (.error.reason // .error.type) elif (.error|type) == "string" then .error else empty end) // empty | tostring | gsub("[\t\r\n]+"; " ")' <<<"$body" 2>/dev/null)
-    reason="HTTP ${http_code:-000}${reason:+: ${reason}}"
   fi
 
   # curl above ran in parallel; serialize just this block write so concurrent jobs never interleave.
@@ -57,7 +34,7 @@ put_policy() {
     flock 200
     printf 'Setting up %s policy...\n%s\n\n' "${desc}" "${result}"
     if (( rc != 0 )); then
-      printf '%s\t%s\n' "${policyname}" "${reason}" >>"$ILM_FAIL_FILE"
+      printf '%s\n' "${policyname}" >>"$ILM_FAIL_FILE"
     fi
   } 200>>"${ILM_OUTPUT_LOCK}"
 
@@ -105,15 +82,10 @@ wait || true
 
 if [[ -s "$ILM_FAIL_FILE" ]]; then
   echo "ERROR: Failed to load ILM policy(s):"
-  while IFS=$'\t' read -r POLICY REASON; do
+  while read -r POLICY; do
-    if [[ -n "$REASON" ]]; then
+    echo " - $POLICY"
-      echo " - ${POLICY}: ${REASON}"
-    else
-      echo " - ${POLICY}"
-    fi
   done < "$ILM_FAIL_FILE"
   exit 1
 else
   echo "Successfully loaded all ILM policies."
-  touch "$ILM_STATEFILE"
 fi

salt/manager/tools/sbin/so-push-drainer

@@ -16,12 +16,9 @@ For each intent file under /opt/so/state/push_pending/*.json whose last_touch
 is older than debounce_seconds, this script:
   * concatenates the actions lists from every ready intent
   * dedupes by (state or __highstate__, tgt, tgt_type)
-  * drops actions whose target matches no accepted minion (e.g. a role with no nodes yet)
   * dispatches a single `salt-run state.orchestrate orch.push_batch --async`
     with the deduped actions list passed as pillar kwargs
   * deletes the contributed intent files on successful dispatch
-  * records the orchestration jid under /opt/so/state/push_dispatched and, on
-    later passes, looks up its result and logs success or per-minion failures
 
 Reactor sls files (push_files, push_pillar) write intents
 but never dispatch directly
@@ -33,36 +30,18 @@ import json
 import logging
 import logging.handlers
 import os
-import re
 import subprocess
 import sys
 import time
 
 import salt.client
-import salt.config
-import salt.utils.minions
 
 PENDING_DIR = '/opt/so/state/push_pending'
 LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
 LOG_FILE = '/opt/so/log/salt/so-push-drainer.log'
-MASTER_CONFIG = '/etc/salt/master'
-
-DISPATCHED_DIR = '/opt/so/state/push_dispatched'
 
 HIGHSTATE_SENTINEL = '__highstate__'
 
-RESULT_CHECK_DELAY = 30
-RESULT_RECHECK_MAX = 300
-RESULT_MAX_AGE = 7200
-RESULT_CHECKS_PER_PASS = 5
-TEXT_LIMIT = 500
-
-# Lead-in salt puts on the comment of an orchestration step that raised.
-STEP_RAISED = 'An exception occurred in this state:'
-
-# salt-run --async reports the jid only in a log line (stderr by default).
-JID_RE = re.compile(r'salt/run/(\d{20})')
-
 
 def _make_logger():
     logger = logging.getLogger('so-push-drainer')
@@ -117,29 +96,6 @@ def _dedupe_actions(actions):
     return deduped
 
 
-def _targeted(actions, log):
-    """Drops actions whose target matches no accepted minion (e.g. a role with no nodes yet), which
-    the orchestration would otherwise report as failed with 'No minions responded'."""
-    try:
-        ckminions = salt.utils.minions.CkMinions(salt.config.master_config(MASTER_CONFIG))
-    except Exception:
-        log.exception('cannot load master config; dispatching without checking targets')
-        return actions
-    targeted = []
-    for action in actions:
-        try:
-            matched = ckminions.check_minions(action['tgt'], action.get('tgt_type', 'compound')).get('minions')
-        except Exception:
-            log.exception('cannot resolve tgt=%s; dispatching it anyway', action.get('tgt'))
-            matched = True
-        if matched:
-            targeted.append(action)
-        else:
-            log.info('skipping %s: no minions match tgt=%s',
-                     'highstate' if action.get('highstate') else action.get('state'), action.get('tgt'))
-    return targeted
-
-
 def _dispatch(actions, log):
     pillar_arg = json.dumps({'actions': actions})
     cmd = [
@@ -157,189 +113,14 @@ def _dispatch(actions, log):
     except subprocess.CalledProcessError as exc:
         log.error('dispatch failed (rc=%s): stdout=%s stderr=%s',
                   exc.returncode, exc.stdout, exc.stderr)
-        return None
+        return False
     except subprocess.TimeoutExpired:
         log.error('dispatch timed out after 60s')
-        return None
+        return False
     except Exception:
         log.exception('dispatch raised')
-        return None
-    output = '{}\n{}'.format(result.stderr or '', result.stdout or '')
-    match = JID_RE.search(output)
-    if not match:
-        log.warning('dispatch accepted but no jid found, result will not be tracked: output=%s',
-                    _trim(output))
-        return ''
-    log.info('dispatch accepted: jid=%s', match.group(1))
-    return match.group(1)
-
-
-def _trim(value):
-    text = value if isinstance(value, str) else json.dumps(value, default=str)
-    lines = [line.strip() for line in text.splitlines() if line.strip()]
-    if 'Traceback (most recent call last):' in text:
-        # Keep the lead-in and the raised exception; the frames are noise in a log line.
-        lines = [text.split('Traceback (most recent call last):', 1)[0].strip(), lines[-1]]
-    text = ' '.join(line for line in lines if line)
-    return text if len(text) <= TEXT_LIMIT else text[:TEXT_LIMIT] + '...'
-
-
-def _unlink(path, log):
-    try:
-        os.unlink(path)
-    except FileNotFoundError:
-        pass
-    except OSError:
-        log.exception('failed to remove %s', path)
-
-
-def _write_record(path, record, log):
-    try:
-        os.makedirs(DISPATCHED_DIR, exist_ok=True)
-        tmp_path = path + '.tmp'
-        with open(tmp_path, 'w') as f:
-            json.dump(record, f)
-        os.rename(tmp_path, path)
-    except Exception:
-        log.exception('failed to record dispatch %s', record.get('jid'))
-
-
-def _record_dispatch(jid, actions, paths, log):
-    record = {'jid': jid, 'dispatched_at': time.time(), 'actions': actions, 'paths': paths}
-    _write_record(os.path.join(DISPATCHED_DIR, '{}.json'.format(jid)), record, log)
-
-
-def _lookup_jid(jid, log):
-    """Returns the job cache entry for jid, {} while it is still running, or None on error."""
-    cmd = ['salt-run', 'jobs.lookup_jid', jid, '--out=json']
-    try:
-        result = subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=60)
-        return json.loads(result.stdout or '{}')
-    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, ValueError) as exc:
-        log.warning('lookup of jid %s failed: %s', jid, exc)
-        return None
-
-
-def _minion_failure(minion_ret):
-    if isinstance(minion_ret, dict):
-        return '; '.join(
-            '{}: {}'.format(state.get('__id__', state_key), _trim(state.get('comment', '')))
-            for state_key, state in minion_ret.items()
-            if isinstance(state, dict) and state.get('result') is False
-        )
-    # A state run rejected before it starts (e.g. another state run is in
-    # progress) returns a list of error strings instead of state results.
-    if isinstance(minion_ret, (list, str)):
-        return _trim(minion_ret)
-    return ''
-
-
-def _step_failures(step):
-    if not isinstance(step, dict) or step.get('result') is not False:
-        return []
-    failures = ['{}: {}'.format(step.get('__id__', step.get('name')), _trim(step.get('comment', '')))]
-    changes = step.get('changes')
-    minion_rets = changes.get('ret') if isinstance(changes, dict) else None
-    if isinstance(minion_rets, dict):
-        for minion, minion_ret in minion_rets.items():
-            text = _minion_failure(minion_ret)
-            if text:
-                failures.append('{}: {}'.format(minion, text))
-    return failures
-
-
-def _orch_failures(ret):
-    if not isinstance(ret, dict):
-        return [_trim(ret)]
-    failures = []
-    for job in ret.values():
-        if not isinstance(job, dict):
-            continue
-        job_ret = job.get('return')
-        data = job_ret.get('data') if isinstance(job_ret, dict) else {}
-        if not isinstance(data, dict):
-            if data:
-                failures.append(_trim(data))
-            data = {}
-        for steps in data.values():
-            if not isinstance(steps, dict):
-                failures.append(_trim(steps))
-                continue
-            for step in steps.values():
-                failures.extend(_step_failures(step))
-        if job.get('success') is False and not failures:
-            failures.append('orchestration reported failure: {}'.format(_trim(job.get('return'))))
-    return failures
-
-
-def _failed_steps(ret):
-    steps = []
-    for job in ret.values() if isinstance(ret, dict) else []:
-        job_ret = job.get('return') if isinstance(job, dict) else None
-        data = job_ret.get('data') if isinstance(job_ret, dict) else None
-        for group in data.values() if isinstance(data, dict) else []:
-            if isinstance(group, dict):
-                steps.extend(step for step in group.values() if isinstance(step, dict) and step.get('result') is False)
-    return steps
-
-
-def _result_unknown(ret):
-    # A step that raised (e.g. salt-master restarted while it waited on a queued state run)
-    # never collected the minion's return, so the state run may still have completed.
-    steps = _failed_steps(ret)
-    return bool(steps) and all(str(step.get('comment', '')).startswith(STEP_RAISED) for step in steps)
-
-
-def _recheck_delay(age):
-    return min(RESULT_RECHECK_MAX, max(RESULT_CHECK_DELAY, age / 4))
-
-
-def _check_dispatched(log, now):
-    due = []
-    for path in glob.glob(os.path.join(DISPATCHED_DIR, '*.json')):
-        record = _read_intent(path, log)
-        if not isinstance(record, dict) or not record.get('jid'):
-            _unlink(path, log)
-            continue
-        age = now - record.get('dispatched_at', 0)
-        last_check = record.get('checked_at', record.get('dispatched_at', 0))
-        if now - last_check >= _recheck_delay(age):
-            due.append((last_check, path, record, age))
-    # Least recently checked first, so pushes that are still running can't starve finished ones.
-    for _, path, record, age in sorted(due, key=lambda item: item[:2])[:RESULT_CHECKS_PER_PASS]:
-        jid = record['jid']
-        try:
-            if _report_result(record, age, log):
-                _unlink(path, log)
-            else:
-                record['checked_at'] = now
-                _write_record(path, record, log)
-        except Exception:
-            # Drop the record so one unreadable result can't fail every pass ahead of the drain.
-            log.exception('cannot evaluate result for jid=%s; no longer tracking', jid)
-            _unlink(path, log)
-
-
-def _report_result(record, age, log):
-    """Logs the outcome of a dispatched push. Returns True once the record is finished with."""
-    jid = record['jid']
-    paths = record.get('paths', [])
-    ret = _lookup_jid(jid, log)
-    if not ret:
-        if age > RESULT_MAX_AGE:
-            log.warning('no result for jid=%s after %ds, no longer tracking; paths=%s', jid, age, paths)
-            return True
         return False
-    failures = _orch_failures(ret)
+    log.info('dispatch accepted: %s', (result.stdout or '').strip())
-    if failures and _result_unknown(ret):
-        log.warning('push result unknown jid=%s paths=%s; the orchestration lost track of the state run, which may '
-                    'still have completed (if not, the change will be applied at the next scheduled highstate): %s',
-                    jid, paths, ' | '.join(failures))
-    elif failures:
-        log.error('push failed jid=%s paths=%s; change will be applied at the next scheduled highstate: %s',
-                  jid, paths, ' | '.join(failures))
-    else:
-        log.info('push succeeded jid=%s paths=%s', jid, paths)
     return True
 
 
@@ -362,9 +143,6 @@ def main():
 
     debounce_seconds = int(push.get('debounce_seconds', 30))
 
-    # Outside the lock: lookups are slow and the reactors take the same lock.
-    _check_dispatched(log, time.time())
-
     os.makedirs(PENDING_DIR, exist_ok=True)
     lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
     try:
@@ -430,20 +208,10 @@ def main():
             len(ready), len(deduped), len(combined_actions),
             debounce_duration, all_paths[:20],
         )
-        for action in deduped:
-            log.info('action: %s tgt=%s', 'highstate' if action.get('highstate') else action.get('state'),
-                     action.get('tgt'))
 
-        targeted = _targeted(deduped, log)
+        if not _dispatch(deduped, log):
-        if targeted:
+            log.warning('dispatch failed; leaving intent files in place for retry')
-            jid = _dispatch(targeted, log)
+            return 1
-            if jid is None:
-                log.warning('dispatch failed; leaving intent files in place for retry')
-                return 1
-            if jid:
-                _record_dispatch(jid, targeted, all_paths[:20], log)
-        else:
-            log.info('no action matches a minion; nothing to dispatch')
 
         for path, _ in ready:
             try:

salt/manager/tools/sbin/so-push-drainer_test.py

@@ -1,569 +0,0 @@
-# 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.
-
-import importlib.util
-import json
-import logging
-import os
-import shutil
-import subprocess
-import sys
-import tempfile
-import time
-import unittest
-from importlib.machinery import SourceFileLoader
-from unittest.mock import MagicMock, patch
-
-HERE = os.path.dirname(os.path.abspath(__file__))
-SCRIPT = os.path.join(HERE, 'so-push-drainer')
-_loader = SourceFileLoader('so_push_drainer', SCRIPT)
-_spec = importlib.util.spec_from_loader('so_push_drainer', _loader)
-drainer = importlib.util.module_from_spec(_spec)
-
-# salt is not installed where these tests run; the drainer only needs salt.client.Caller and CkMinions.
-# Mocked only while the drainer loads: run from the repo root, 'salt' is this repo's salt/ directory.
-_salt = MagicMock()
-with patch.dict(sys.modules, {'salt': _salt, 'salt.client': _salt.client, 'salt.config': _salt.config,
-                              'salt.utils': _salt.utils, 'salt.utils.minions': _salt.utils.minions}):
-    _loader.exec_module(drainer)
-
-MASTER = 'manager.localdomain_master'
-JID = '20260930171554259426'
-ASYNC_STDERR = ('[WARNING ] Running in asynchronous mode. Results of this execution may be collected '
-                'by attaching to the master event bus or by examining the master job cache, if '
-                'configured. This execution is running under tag salt/run/{}\n'.format(JID))
-CONFLICT = ('The function "state.sls" is running as PID 372218 and was started at '
-            '2026, Sep 30 17:15:40.466233 with jid 20260930171540466233')
-
-
-def _orch_ret(steps, success=True):
-    return {MASTER: {
-        'fun': 'runner.state.orchestrate',
-        'jid': JID,
-        'return': {'data': {MASTER: steps}, 'outputter': 'highstate', 'retcode': 0 if success else 1},
-        'success': success,
-    }}
-
-
-REFRESH_STEP = {
-    'salt_|-refresh_pillar_1_|-saltutil.refresh_pillar_|-function': {
-        '__id__': 'refresh_pillar_1', 'result': True,
-        'changes': {'ret': {'manager_standalone': True}},
-        'comment': 'Function ran successfully.',
-    },
-}
-
-CONFLICT_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
-        '__id__': 'apply_soc_1', 'result': False,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': [CONFLICT]}},
-        'comment': 'Run failed on minions: manager_standalone',
-    },
-}), success=False)
-
-STATE_FAIL_RET = _orch_ret({
-    'salt_|-apply_hydra_1_|-apply_hydra_1_|-state': {
-        '__id__': 'apply_hydra_1', 'result': False,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': {
-            'test_|-no_license_|-no_license_|-fail_without_changes': {
-                '__id__': 'hydra.enabled_no_license_detected', 'result': False,
-                'comment': 'This is a feature supported only for customers with a valid license.',
-            },
-            'file_|-hydra_conf_|-/opt/so/conf/hydra_|-managed': {'result': True, 'comment': 'ok'},
-        }}},
-        'comment': 'Run failed on minions: manager_standalone',
-    },
-}, success=False)
-
-RAISED = ('An exception occurred in this state: Traceback (most recent call last):\n'
-          '  File "salt/client/__init__.py", line 1934, in pub\n'
-          '    raise AuthenticationError(err_msg)\n'
-          'salt.exceptions.AuthenticationError: Authentication error occurred.\n')
-
-RAISED_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_hydra_1_|-apply_hydra_1_|-state': {
-        '__id__': 'apply_hydra_1', 'result': False, 'changes': {}, 'comment': RAISED,
-    },
-}), success=False)
-
-SUCCESS_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_telegraf_1_|-apply_telegraf_1_|-state': {
-        '__id__': 'apply_telegraf_1', 'result': True,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': {
-            'file_|-tgrafconf_|-/opt/so/conf/telegraf/etc/telegraf.conf_|-managed': {'result': True},
-        }}},
-        'comment': 'States ran successfully.',
-    },
-}))
-
-
-class DrainerTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.tmpdir = tempfile.mkdtemp()
-        self.pending = os.path.join(self.tmpdir, 'push_pending')
-        self.dispatched = os.path.join(self.tmpdir, 'push_dispatched')
-        os.makedirs(self.pending)
-        for name, value in (
-            ('PENDING_DIR', self.pending),
-            ('LOCK_FILE', os.path.join(self.pending, '.lock')),
-            ('DISPATCHED_DIR', self.dispatched),
-            ('LOG_FILE', os.path.join(self.tmpdir, 'log', 'so-push-drainer.log')),
-        ):
-            patcher = patch.object(drainer, name, value)
-            patcher.start()
-            self.addCleanup(patcher.stop)
-        self.log = MagicMock()
-
-    def tearDown(self):
-        shutil.rmtree(self.tmpdir, ignore_errors=True)
-
-    def write_json(self, directory, name, data):
-        os.makedirs(directory, exist_ok=True)
-        path = os.path.join(directory, name)
-        with open(path, 'w') as f:
-            if isinstance(data, str):
-                f.write(data)
-            else:
-                json.dump(data, f)
-        return path
-
-    def logged(self, level):
-        return ' '.join(c.args[0] % c.args[1:] for c in getattr(self.log, level).call_args_list)
-
-
-class TestHelpers(DrainerTestCase):
-
-    def test_make_logger_adds_handler_once(self):
-        logger = logging.getLogger('so-push-drainer')
-
-        def close_handlers():
-            for handler in logger.handlers:
-                handler.close()
-            logger.handlers.clear()
-
-        self.addCleanup(close_handlers)
-        logger.handlers.clear()
-        self.assertIs(drainer._make_logger(), logger)
-        drainer._make_logger()
-        self.assertEqual(len(logger.handlers), 1)
-        self.assertTrue(os.path.isdir(os.path.dirname(drainer.LOG_FILE)))
-
-    def test_load_push_cfg(self):
-        with patch.object(drainer.salt.client, 'Caller') as caller:
-            caller.return_value.cmd.return_value = {'enabled': False}
-            self.assertEqual(drainer._load_push_cfg(), {'enabled': False})
-            caller.return_value.cmd.return_value = 'garbage'
-            self.assertEqual(drainer._load_push_cfg(), {})
-
-    def test_read_intent(self):
-        good = self.write_json(self.pending, 'good.json', {'a': 1})
-        bad = self.write_json(self.pending, 'bad.json', '{nope')
-        self.assertEqual(drainer._read_intent(good, self.log), {'a': 1})
-        self.assertIsNone(drainer._read_intent(bad, self.log))
-        with patch('builtins.open', side_effect=RuntimeError('boom')):
-            self.assertIsNone(drainer._read_intent(good, self.log))
-        self.log.exception.assert_called_once()
-
-    def test_dedupe_actions(self):
-        actions = [
-            'not a dict',
-            {'state': 'soc'},
-            {'state': 'soc', 'tgt': '*'},
-            {'state': 'soc', 'tgt': '*', 'tgt_type': 'compound'},
-            {'highstate': True, 'tgt': '*'},
-            {'state': 'soc', 'tgt': 'node1', 'tgt_type': 'glob'},
-        ]
-        self.assertEqual(drainer._dedupe_actions(actions), [actions[2], actions[4], actions[5]])
-
-    def test_trim(self):
-        self.assertEqual(drainer._trim('  text \n'), 'text')
-        self.assertEqual(drainer._trim(['a']), '["a"]')
-        self.assertEqual(drainer._trim(None), 'null')
-        self.assertEqual(drainer._trim('x' * 600), 'x' * drainer.TEXT_LIMIT + '...')
-
-    def test_trim_traceback(self):
-        comment = ('An exception occurred in this state: Traceback (most recent call last):\n'
-                   '  File "salt/client/__init__.py", line 1934, in pub\n'
-                   '    raise AuthenticationError(err_msg)\n'
-                   'salt.exceptions.AuthenticationError: Authentication error occurred.\n')
-        self.assertEqual(drainer._trim(comment), 'An exception occurred in this state: '
-                         'salt.exceptions.AuthenticationError: Authentication error occurred.')
-        self.assertEqual(drainer._trim('line one\n  line two\n'), 'line one line two')
-
-    def test_unlink(self):
-        drainer._unlink(os.path.join(self.tmpdir, 'missing'), self.log)
-        self.log.exception.assert_not_called()
-        drainer._unlink(self.tmpdir, self.log)
-        self.log.exception.assert_called_once()
-
-
-class TestDispatch(DrainerTestCase):
-
-    def run_dispatch(self, **kwargs):
-        with patch.object(drainer.subprocess, 'run', **kwargs) as run:
-            jid = drainer._dispatch([{'state': 'soc', 'tgt': '*'}], self.log)
-        return jid, run
-
-    def test_jid_parsed_from_stderr(self):
-        jid, run = self.run_dispatch(return_value=MagicMock(stdout='', stderr=ASYNC_STDERR))
-        self.assertEqual(jid, JID)
-        cmd = run.call_args[0][0]
-        self.assertEqual(cmd[:3], ['salt-run', 'state.orchestrate', 'orch.push_batch'])
-        self.assertIn('--async', cmd)
-
-    def test_jid_parsed_from_stdout(self):
-        jid, _ = self.run_dispatch(return_value=MagicMock(stdout=ASYNC_STDERR, stderr=None))
-        self.assertEqual(jid, JID)
-
-    def test_no_jid(self):
-        jid, _ = self.run_dispatch(return_value=MagicMock(stdout='unexpected output', stderr=None))
-        self.assertEqual(jid, '')
-        self.assertIn('output=unexpected output', self.logged('warning'))
-
-    def test_failures_return_none(self):
-        for exc in (subprocess.CalledProcessError(1, 'salt-run', 'out', 'err'),
-                    subprocess.TimeoutExpired('salt-run', 60),
-                    RuntimeError('boom')):
-            jid, _ = self.run_dispatch(side_effect=exc)
-            self.assertIsNone(jid)
-
-    def test_record_dispatch(self):
-        drainer._record_dispatch(JID, [{'state': 'soc'}], ['audit:soc.config.licenseKey'], self.log)
-        with open(os.path.join(self.dispatched, JID + '.json')) as f:
-            record = json.load(f)
-        self.assertEqual(record['jid'], JID)
-        self.assertEqual(record['paths'], ['audit:soc.config.licenseKey'])
-        self.assertIn('dispatched_at', record)
-
-    def test_record_dispatch_errors(self):
-        with patch.object(drainer.os, 'makedirs', side_effect=OSError('ro')):
-            drainer._record_dispatch(JID, [], [], self.log)
-        drainer._record_dispatch(JID, [object()], [], self.log)
-        self.assertEqual(self.log.exception.call_count, 2)
-        self.assertFalse(os.path.exists(os.path.join(self.dispatched, JID + '.json')))
-
-
-class TestResults(DrainerTestCase):
-
-    def test_lookup_jid(self):
-        with patch.object(drainer.subprocess, 'run') as run:
-            run.return_value = MagicMock(stdout=json.dumps(SUCCESS_RET))
-            self.assertEqual(drainer._lookup_jid(JID, self.log), SUCCESS_RET)
-            self.assertEqual(run.call_args[0][0], ['salt-run', 'jobs.lookup_jid', JID, '--out=json'])
-            run.return_value = MagicMock(stdout='')
-            self.assertEqual(drainer._lookup_jid(JID, self.log), {})
-            run.return_value = MagicMock(stdout='not json')
-            self.assertIsNone(drainer._lookup_jid(JID, self.log))
-            run.side_effect = subprocess.TimeoutExpired('salt-run', 60)
-            self.assertIsNone(drainer._lookup_jid(JID, self.log))
-
-    def test_minion_failure_shapes(self):
-        self.assertEqual(drainer._minion_failure([CONFLICT]), json.dumps([CONFLICT]))
-        self.assertEqual(drainer._minion_failure('Rendering SLS failed'), 'Rendering SLS failed')
-        self.assertEqual(drainer._minion_failure(True), '')
-        self.assertEqual(drainer._minion_failure({'a': {'result': True}}), '')
-
-    def test_orch_failures_conflict(self):
-        failures = drainer._orch_failures(CONFLICT_RET)
-        self.assertEqual(failures[0], 'apply_soc_1: Run failed on minions: manager_standalone')
-        self.assertIn('manager_standalone', failures[1])
-        self.assertIn('is running as PID 372218', failures[1])
-        self.assertEqual(len(failures), 2)
-
-    def test_orch_failures_failed_state(self):
-        failures = drainer._orch_failures(STATE_FAIL_RET)
-        self.assertEqual(len(failures), 2)
-        self.assertIn('hydra.enabled_no_license_detected: This is a feature', failures[1])
-        self.assertNotIn('hydra_conf', failures[1])
-
-    def test_orch_failures_success(self):
-        self.assertEqual(drainer._orch_failures(SUCCESS_RET), [])
-
-    def test_orch_failures_render_error(self):
-        ret = {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS failed"]'])
-
-    def test_orch_failures_not_a_dict(self):
-        self.assertEqual(drainer._orch_failures(['No minions matched']), ['["No minions matched"]'])
-        self.assertEqual(drainer._orch_failures('Runner error'), ['Runner error'])
-
-    def test_orch_failures_data_not_a_dict(self):
-        ret = {MASTER: {'return': {'data': ["Rendering SLS 'orch.push_batch' failed"]}, 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS \'orch.push_batch\' failed"]'])
-
-    def test_orch_failures_odd_changes(self):
-        for changes in ('Run failed', {'ret': ['manager_standalone']}):
-            ret = _orch_ret({'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
-                '__id__': 'apply_soc_1', 'result': False, 'changes': changes, 'comment': 'Run failed on minions',
-            }}, success=False)
-            self.assertEqual(drainer._orch_failures(ret), ['apply_soc_1: Run failed on minions'])
-
-    def test_orch_failures_unparsed(self):
-        self.assertEqual(drainer._orch_failures({MASTER: 'odd'}), [])
-        ret = {MASTER: {'return': 'Exception occurred', 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['orchestration reported failure: Exception occurred'])
-
-    def test_result_unknown(self):
-        self.assertTrue(drainer._result_unknown(RAISED_RET))
-        for ret in (CONFLICT_RET, STATE_FAIL_RET, SUCCESS_RET, ['No minions matched'], {MASTER: 'odd'},
-                    {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}):
-            self.assertFalse(drainer._result_unknown(ret), ret)
-        mixed = _orch_ret(dict(RAISED_RET[MASTER]['return']['data'][MASTER],
-                               **CONFLICT_RET[MASTER]['return']['data'][MASTER]), success=False)
-        self.assertFalse(drainer._result_unknown(mixed))
-
-    def record(self, jid, age, now):
-        return self.write_json(self.dispatched, jid + '.json', {
-            'jid': jid, 'dispatched_at': now - age, 'actions': [], 'paths': ['audit:' + jid],
-        })
-
-    def test_check_dispatched(self):
-        now = time.time()
-        results = {
-            '1_failed': CONFLICT_RET,
-            '2_ok': SUCCESS_RET,
-            '3_pending': {},
-            '4_expired': None,
-            '6_unknown': RAISED_RET,
-        }
-        young = self.record('0_young', 5, now)
-        paths = {jid: self.record(jid, 60, now) for jid in results}
-        paths['4_expired'] = self.record('4_expired', drainer.RESULT_MAX_AGE + 1, now)
-        bad = self.write_json(self.dispatched, '5_bad.json', '{nope')
-        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: results[jid]):
-            drainer._check_dispatched(self.log, now)
-
-        self.assertTrue(os.path.exists(young))
-        with open(paths['3_pending']) as f:
-            self.assertEqual(json.load(f)['checked_at'], now)
-        for jid in ('1_failed', '2_ok', '4_expired', '6_unknown'):
-            self.assertFalse(os.path.exists(paths[jid]), jid)
-        self.assertFalse(os.path.exists(bad))
-        self.assertIn('push failed jid=1_failed', self.logged('error'))
-        self.assertIn('is running as PID 372218', self.logged('error'))
-        self.assertIn('push succeeded jid=2_ok', self.logged('info'))
-        self.assertIn('no result for jid=4_expired', self.logged('warning'))
-        self.assertIn('push result unknown jid=6_unknown', self.logged('warning'))
-        self.assertIn('AuthenticationError: Authentication error occurred.', self.logged('warning'))
-        self.assertNotIn('6_unknown', self.logged('error'))
-
-    def test_check_dispatched_survives_bad_result(self):
-        now = time.time()
-        bad = self.record('1_bad', 60, now)
-        good = self.record('2_ok', 60, now)
-
-        def orch_failures(ret):
-            if ret == 'boom':
-                raise ValueError('unexpected shape')
-            return []
-
-        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: 'boom' if jid == '1_bad' else SUCCESS_RET), \
-                patch.object(drainer, '_orch_failures', side_effect=orch_failures):
-            drainer._check_dispatched(self.log, now)
-        self.assertFalse(os.path.exists(bad))
-        self.assertFalse(os.path.exists(good))
-        self.log.exception.assert_called_once()
-        self.assertIn('jid=1_bad', self.log.exception.call_args[0][0] % self.log.exception.call_args[0][1:])
-        self.assertIn('push succeeded jid=2_ok', self.logged('info'))
-
-    def test_recheck_delay(self):
-        self.assertEqual(drainer._recheck_delay(10), drainer.RESULT_CHECK_DELAY)
-        self.assertEqual(drainer._recheck_delay(400), 100)
-        self.assertEqual(drainer._recheck_delay(drainer.RESULT_MAX_AGE), drainer.RESULT_RECHECK_MAX)
-
-    def test_check_dispatched_limit_rotates(self):
-        now = time.time()
-        limit = drainer.RESULT_CHECKS_PER_PASS
-        jids = ['{:02d}'.format(i) for i in range(limit + 2)]
-        for jid in jids:
-            self.record(jid, 60, now)
-        with patch.object(drainer, '_lookup_jid', return_value={}) as lookup:
-            drainer._check_dispatched(self.log, now)
-            self.assertEqual([c.args[0] for c in lookup.call_args_list], jids[:limit])
-            lookup.reset_mock()
-            drainer._check_dispatched(self.log, now + 40)
-        self.assertEqual([c.args[0] for c in lookup.call_args_list], jids[limit:] + jids[:limit - 2])
-
-    def test_check_dispatched_not_blocked_by_running(self):
-        now = time.time()
-        for i in range(drainer.RESULT_CHECKS_PER_PASS):
-            self.record('1_running{}'.format(i), 600, now)
-        done = self.record('2_done', 60, now)
-
-        def lookup(jid, log):
-            return SUCCESS_RET if jid == '2_done' else {}
-
-        with patch.object(drainer, '_lookup_jid', side_effect=lookup) as lookup_jid:
-            drainer._check_dispatched(self.log, now)
-            self.assertTrue(os.path.exists(done))
-            lookup_jid.reset_mock()
-            drainer._check_dispatched(self.log, now + 15)
-        self.assertEqual([c.args[0] for c in lookup_jid.call_args_list], ['2_done'])
-        self.assertFalse(os.path.exists(done))
-        self.assertIn('push succeeded jid=2_done', self.logged('info'))
-
-
-class TestTargeted(DrainerTestCase):
-
-    def setUp(self):
-        super().setUp()
-        self.ckminions = MagicMock()
-        patcher = patch.object(drainer.salt.utils.minions, 'CkMinions', return_value=self.ckminions)
-        self.CkMinions = patcher.start()
-        self.addCleanup(patcher.stop)
-
-    def test_drops_unmatched(self):
-        matches = {'G@role:so-sensor': ['sensor_sensor'], 'manager_manager': ['manager_manager']}
-        self.ckminions.check_minions.side_effect = lambda tgt, tgt_type: {'minions': matches.get(tgt, [])}
-        actions = [
-            {'state': 'strelka', 'tgt': 'G@role:so-sensor'},
-            {'state': 'suricata', 'tgt': 'G@role:so-heavynode'},
-            {'highstate': True, 'tgt': 'G@role:so-idh'},
-            {'state': 'soc', 'tgt': 'manager_manager', 'tgt_type': 'list'},
-        ]
-        self.assertEqual(drainer._targeted(actions, self.log), [actions[0], actions[3]])
-        self.CkMinions.assert_called_once()
-        self.assertEqual([c.args for c in self.ckminions.check_minions.call_args_list], [
-            ('G@role:so-sensor', 'compound'), ('G@role:so-heavynode', 'compound'),
-            ('G@role:so-idh', 'compound'), ('manager_manager', 'list'),
-        ])
-        info = self.logged('info')
-        self.assertIn('skipping suricata: no minions match tgt=G@role:so-heavynode', info)
-        self.assertIn('skipping highstate: no minions match tgt=G@role:so-idh', info)
-        self.assertNotIn('strelka', info)
-
-    def test_master_config_error_keeps_actions(self):
-        actions = [{'state': 'soc', 'tgt': '*'}]
-        with patch.object(drainer.salt.config, 'master_config', side_effect=OSError('no config')):
-            self.assertEqual(drainer._targeted(actions, self.log), actions)
-        self.log.exception.assert_called_once()
-        self.CkMinions.assert_not_called()
-
-    def test_check_error_keeps_action(self):
-        actions = [{'state': 'soc', 'tgt': 'G@role:'}]
-        self.ckminions.check_minions.side_effect = RuntimeError('bad target')
-        self.assertEqual(drainer._targeted(actions, self.log), actions)
-        self.log.exception.assert_called_once()
-
-
-class TestMain(DrainerTestCase):
-
-    def setUp(self):
-        super().setUp()
-        self.cfg = {'enabled': True, 'debounce_seconds': 30}
-        for name, kwargs in (
-            ('_make_logger', {'return_value': self.log}),
-            ('_load_push_cfg', {'side_effect': lambda: self.cfg}),
-            ('_check_dispatched', {}),
-            ('_targeted', {'side_effect': lambda actions, log: actions}),
-        ):
-            patcher = patch.object(drainer, name, **kwargs)
-            setattr(self, name, patcher.start())
-            self.addCleanup(patcher.stop)
-
-    def intent(self, name, age=60, actions=None, paths=None):
-        now = time.time()
-        return self.write_json(self.pending, name, {
-            'first_touch': now - age - 5, 'last_touch': now - age,
-            'actions': [{'state': 'soc', 'tgt': '*'}] if actions is None else actions,
-            'paths': paths or ['audit:soc.config.licenseKey'],
-        })
-
-    def test_no_pending_dir(self):
-        shutil.rmtree(self.pending)
-        self.assertEqual(drainer.main(), 0)
-        self._load_push_cfg.assert_not_called()
-
-    def test_cfg_error(self):
-        self._load_push_cfg.side_effect = RuntimeError('no salt')
-        self.assertEqual(drainer.main(), 1)
-
-    def test_disabled(self):
-        self.cfg['enabled'] = False
-        self.assertEqual(drainer.main(), 0)
-        self._check_dispatched.assert_not_called()
-
-    def test_no_intents_still_checks_results(self):
-        self.assertEqual(drainer.main(), 0)
-        self._check_dispatched.assert_called_once()
-
-    def test_debounce_and_broken(self):
-        young = self.intent('young.json', age=1)
-        broken = self.write_json(self.pending, 'broken.json', '{nope')
-        with patch.object(drainer, '_dispatch') as dispatch:
-            self.assertEqual(drainer.main(), 0)
-        dispatch.assert_not_called()
-        self.assertTrue(os.path.exists(young))
-        self.assertFalse(os.path.exists(broken))
-
-    def test_broken_unlink_error_ignored(self):
-        self.write_json(self.pending, 'broken.json', '{nope')
-        with patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-
-    def test_no_usable_actions(self):
-        path = self.intent('empty.json', actions=[{'state': 'soc'}])
-        self.assertEqual(drainer.main(), 0)
-        self.assertFalse(os.path.exists(path))
-        self.intent('empty.json', actions=[{'state': 'soc'}])
-        with patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-
-    def test_dispatch_failure_keeps_intents(self):
-        path = self.intent('pillar_soc.json')
-        with patch.object(drainer, '_dispatch', return_value=None):
-            self.assertEqual(drainer.main(), 1)
-        self.assertTrue(os.path.exists(path))
-
-    def test_dispatch_records_jid(self):
-        soc = self.intent('pillar_soc.json')
-        hs = self.intent('pillar_global.json', actions=[{'highstate': True, 'tgt': '*'}], paths=['audit:global.x'])
-        with patch.object(drainer, '_dispatch', return_value=JID) as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        self.assertEqual(len(dispatch.call_args[0][0]), 2)
-        record.assert_called_once()
-        self.assertEqual(record.call_args[0][0], JID)
-        self.assertEqual(sorted(record.call_args[0][2]), ['audit:global.x', 'audit:soc.config.licenseKey'])
-        self.assertFalse(os.path.exists(soc))
-        self.assertFalse(os.path.exists(hs))
-        self.assertIn('action: highstate tgt=*', self.logged('info'))
-
-    def test_dispatch_without_jid_not_recorded(self):
-        self.intent('pillar_soc.json')
-        with patch.object(drainer, '_dispatch', return_value=''), \
-                patch.object(drainer, '_record_dispatch') as record, \
-                patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-        record.assert_not_called()
-        self.log.exception.assert_called_once()
-
-    def test_untargeted_not_dispatched(self):
-        path = self.intent('strelka.json', actions=[{'state': 'strelka', 'tgt': 'G@role:so-sensor'}])
-        self._targeted.side_effect = None
-        self._targeted.return_value = []
-        with patch.object(drainer, '_dispatch') as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        dispatch.assert_not_called()
-        record.assert_not_called()
-        self.assertFalse(os.path.exists(path))
-        self.assertIn('no action matches a minion; nothing to dispatch', self.logged('info'))
-
-    def test_dispatches_only_targeted(self):
-        kept = {'state': 'soc', 'tgt': '*'}
-        self.intent('mixed.json', actions=[kept, {'state': 'strelka', 'tgt': 'G@role:so-sensor'}])
-        self._targeted.side_effect = None
-        self._targeted.return_value = [kept]
-        with patch.object(drainer, '_dispatch', return_value=JID) as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        self.assertEqual(dispatch.call_args[0][0], [kept])
-        self.assertEqual(record.call_args[0][1], [kept])
-
-
-if __name__ == '__main__':
-    unittest.main()

salt/manager/tools/sbin/soup

@@ -121,6 +121,9 @@ check_err() {
       161)
         echo 'Required intermediate Elasticsearch upgrade not complete'
       ;;
+      162)
+        echo 'One or more Elastic Agent nodes do not support the x86-64-v3 CPU instruction set'
+      ;;
       170)
         echo "Intermediate upgrade completed successfully to $next_step_so_version, but next soup to Security Onion $originally_requested_so_version could not be started automatically."
         echo "Start soup again manually to continue the upgrade to Security Onion $originally_requested_so_version."
@@ -348,6 +351,83 @@ check_cluster_health() {
   exit 0
 }
 
+no_soup_for_you() {
+  echo ""
+  echo "No soup for you!"
+  exit 162
+}
+
+check_cpu_compatibility() {
+  # Roles running a container built from the so-elastic-agent image; mirrors the
+  # elasticagent and elasticfleet entries in salt/reactor/pillar_push_map.yaml.
+  local cpu_target='G@role:so-heavynode or G@role:so-eval or G@role:so-fleet or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+  local expected_nodes cpu_results node result confirm
+  local -a unsupported=() offline=()
+
+  echo "Checking that Elastic Agent nodes support the x86-64-v3 CPU instruction set now required by Elastic."
+
+  if [[ "$SKIP_CPU_CHECK" == "true" ]]; then
+    printf "\nSkipping the x86-64-v3 CPU check because --skip-cpu-check was specified.\n\n"
+    return
+  fi
+
+  # Nodes that never answer are absent from the results, so diff against who should have.
+  expected_nodes=$(salt -C "$cpu_target" --preview-target --out=json 2>/dev/null | jq -r '.[]?') || true
+  if [[ -z "$expected_nodes" ]]; then
+    printf "\nCould not determine which nodes run the Elastic Agent, so the x86-64-v3 CPU check cannot run.\n"
+    no_soup_for_you
+  fi
+
+  cpu_results=$(salt -t 30 -C "$cpu_target" cmd.run "/lib64/ld-linux-x86-64.so.2 --help | grep x86-64-v3" --out=json 2>/dev/null) || true
+
+  while IFS= read -r node; do
+    [[ -z "$node" ]] && continue
+    result=$(jq -r --arg node "$node" '.[$node] // empty' <<< "$cpu_results" 2>/dev/null)
+    if [[ -z "$result" || "$result" == *"did not return"* ]]; then
+      offline+=("$node")
+    elif [[ "$result" != *"x86-64-v3 (supported"* ]]; then
+      # glibc appends "(supported, searched)" only when supported; the open paren keeps
+      # this from matching a future "(unsupported".
+      unsupported+=("$node")
+    fi
+  done <<< "$expected_nodes"
+
+  if [[ ${#unsupported[@]} -eq 0 && ${#offline[@]} -eq 0 ]]; then
+    printf "\nAll Elastic Agent nodes support x86-64-v3. We can proceed with SOUP.\n\n"
+    return
+  fi
+
+  echo ""
+  if [[ ${#unsupported[@]} -gt 0 ]]; then
+    echo "The following node(s) do NOT support the x86-64-v3 CPU instruction set:"
+    printf '  %s\n' "${unsupported[@]}"
+    echo ""
+    echo "Upstream Elastic now builds its binaries for x86-64-v3, so these nodes can no"
+    echo "longer run Elastic. Upgrading them WILL BREAK them."
+    echo ""
+  fi
+  if [[ ${#offline[@]} -gt 0 ]]; then
+    echo "The following node(s) did not respond and could not be checked:"
+    printf '  %s\n' "${offline[@]}"
+    echo ""
+    echo "These nodes are offline, so we cannot confirm they support x86-64-v3, which"
+    echo "upstream Elastic now requires."
+    echo ""
+  fi
+
+  if [[ -n $UNATTENDED ]]; then
+    echo "Unattended mode cannot prompt for an override. Re-run soup interactively, or pass --skip-cpu-check to bypass this check."
+    no_soup_for_you
+  fi
+
+  read -rp "Type 'override' to continue anyway, or press Enter to exit: " confirm
+  if [[ "${confirm,,}" == "override" ]]; then
+    printf "\nOverride accepted. Continuing at your own risk.\n\n"
+  else
+    no_soup_for_you
+  fi
+}
+
 check_fleet_server() {
   echo "Checking that Elastic Fleet Server is responding."
   # Modeled on the wait_for_so-elastic-fleet state check in elasticfleet/enabled.sls,
@@ -2088,6 +2168,8 @@ main() {
   echo "Let's see if we need to update Security Onion."
   upgrade_check
 
+  check_cpu_compatibility
+
   upgrade_space
 
   echo "Verifying Elasticsearch version compatibility across the grid before upgrading."
@@ -2366,6 +2448,17 @@ fi
   echo "### soup has been served at $(date) ###"
 }
 
+SKIP_CPU_CHECK=false
+declare -a SOUP_ARGS=()
+for arg in "$@"; do
+  if [[ "$arg" == "--skip-cpu-check" ]]; then
+    SKIP_CPU_CHECK=true
+  else
+    SOUP_ARGS+=("$arg")
+  fi
+done
+set -- "${SOUP_ARGS[@]}"
+
 while getopts ":b:f:y" opt; do
   case ${opt} in
     b )
@@ -2389,7 +2482,7 @@ while getopts ":b:f:y" opt; do
       ISOLOC="$OPTARG"
     ;;
     \? )
-      echo "Usage: soup [-b] [-y] [-f <iso location>]"
+      echo "Usage: soup [-b] [-y] [-f <iso location>] [--skip-cpu-check]"
       exit 1
     ;;
     : )

salt/orch/push_batch.sls

@@ -3,8 +3,6 @@
 {% set BATCH = AUTOAPPLY.batch %}
 {% set BATCH_WAIT = AUTOAPPLY.batch_wait %}
 
-{# queue must be a top-level salt.state arg (kwarg is ignored); an int is max_queue and still fails on conflict #}
-
 {% for action in actions %}
 {%   if action.get('highstate') %}
 apply_highstate_{{ loop.index }}:
@@ -14,7 +12,8 @@ apply_highstate_{{ loop.index }}:
     - highstate: True
     - batch: {{ action.get('batch', BATCH) }}
     - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
-    - queue: True
+    - kwarg:
+        queue: 2
 {%   else %}
 refresh_pillar_{{ loop.index }}:
   salt.function:
@@ -30,7 +29,8 @@ apply_{{ action.state | replace('.', '_') }}_{{ loop.index }}:
       - {{ action.state }}
     - batch: {{ action.get('batch', BATCH) }}
     - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
-    - queue: True
+    - kwarg:
+        queue: 2
     - require:
       - salt: refresh_pillar_{{ loop.index }}
 {%   endif %}

salt/reactor/push_pillar.sls

@@ -7,7 +7,7 @@
 # Two branches, keyed on node_id:
 #   A) node_id populated -> the change is scoped to that one minion. Look up the
 #      app in pillar_push_map.yaml and write an intent that runs the app's mapped
-#      state(s) on just that node, and only if the node is in the app's target.
+#      state(s) targeted to just that node.
 #   B) node_id empty -> grid-wide app change. Look up the app in
 #      pillar_push_map.yaml and write an intent with the entry's actions as-is.
 #
@@ -116,17 +116,15 @@ def _app_from_setting(setting_id):
 
 
 def _node_actions(entry, node_id):
-    # Copy the app's mapped actions but narrow each one to the single node.
+    # Copy the app's mapped actions but retarget each one to the single node.
     # Preserves the state/highstate selection and any batch/batch_wait overrides.
-    # The app's own target is kept so the state is not applied to a node that does not
-    # run the app (e.g. an elasticsearch setting saved for a sensor fails to render there).
     actions = []
     for action in entry:
         if not isinstance(action, dict):
             continue
         node_action = dict(action)
-        node_action['tgt'] = 'L@{} and ( {} )'.format(node_id, action.get('tgt', '*'))
+        node_action['tgt'] = node_id
-        node_action['tgt_type'] = 'compound'
+        node_action['tgt_type'] = 'glob'
         actions.append(node_action)
     return actions
 
@@ -140,7 +138,6 @@ def run():
     # top level so the reactor is robust to either shape.
     event = data.get('data', data)  # noqa: F821 -- data provided by reactor
     setting_id = event.get('setting_id', '')
-    audit_id = event.get('id')
     node_id = (event.get('node_id') or '').strip()
 
     app = _app_from_setting(setting_id)
@@ -153,8 +150,8 @@ def run():
     if not entry:
         LOG.warning(
             'push_pillar: app "%s" is not in pillar_push_map.yaml; change will be '
-            'picked up at the next scheduled highstate (setting_id=%s audit_id=%s)',
+            'picked up at the next scheduled highstate (setting_id=%s)',
-            app, setting_id, audit_id,
+            app, setting_id,
         )
         return {}
 
@@ -168,12 +165,12 @@ def run():
             'node_{}_{}'.format(node_id, app), actions,
             'audit:{}@{}'.format(setting_id, node_id),
         )
-        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s audit_id=%s)',
+        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s)',
-                 app, node_id, setting_id, audit_id)
+                 app, node_id, setting_id)
         return {}
 
     # Branch B: grid-wide app change -> use the map entry's actions as-is.
     actions = list(entry)  # copy to avoid mutating the cache
     _write_intent('pillar_{}'.format(app), actions, 'audit:{}'.format(setting_id))
-    LOG.info('push_pillar: app intent updated for %s (setting_id=%s audit_id=%s)', app, setting_id, audit_id)
+    LOG.info('push_pillar: app intent updated for %s (setting_id=%s)', app, setting_id)
     return {}

setup/so-functions

@@ -976,50 +976,27 @@ docker_seed_update() {
 	set_progress_str "$docker_seed_update_percent" "Downloading $name"
 }
 
-registry_tar_has_soversion() {
-	local tarfile=$1
-	local version=$2
-
-	[ -n "$version" ] || return 1
-	[ -f "$tarfile" ] || return 1
-	# grep -q in a pipe SIGPIPEs tar; under pipefail a match would look like failure.
-	grep -qF "/_manifests/tags/${version}/" < <(tar -tf "$tarfile" 2>/dev/null)
-}
-
 docker_seed_registry() {
 	local VERSION="$SOVERSION"
-	local REGISTRY_TAR=/nsm/docker-registry/docker/registry.tar
 
-	if [ -f "$REGISTRY_TAR" ] && registry_tar_has_soversion "$REGISTRY_TAR" "$VERSION"; then
+	if [ -f /nsm/docker-registry/docker/registry.tar ]; then
-		info "Extracting $REGISTRY_TAR"
+		logCmd "tar xvf /nsm/docker-registry/docker/registry.tar -C /nsm/docker-registry/docker"
-		if ! tar xvf "$REGISTRY_TAR" -C /nsm/docker-registry/docker >> "$setup_log" 2>&1; then
+		logCmd "rm /nsm/docker-registry/docker/registry.tar"
-			error "Failed to extract $REGISTRY_TAR"
-			return 1
-		fi
-		logCmd "rm -f $REGISTRY_TAR"
-		return 0
-	elif [ -f "$REGISTRY_TAR" ]; then
-		info "Not using $REGISTRY_TAR; it has no images tagged '${VERSION}'"
-		if [[ "$is_airgap" == true ]]; then
-			error "Airgap install: $REGISTRY_TAR does not match Security Onion version '${VERSION}'. Use the ISO for this version."
-			return 1
-		fi
 	elif [[ -d /nsm/docker-registry/docker/registry && ( -f /etc/SOCLOUD || "$is_airgap" == true ) ]]; then
 		echo "Using existing docker registry content"
-		return 0
-	fi
-
-	if [ "$install_type" == 'IMPORT' ]; then
-		container_list 'so-import'	
 	else
-		container_list
+		if [ "$install_type" == 'IMPORT' ]; then
-	fi
+			container_list 'so-import'	
+		else
+			container_list
+		fi
 
-	docker_seed_update_percent=25
+		docker_seed_update_percent=25
 
-	update_docker_containers 'netinstall' '' 'docker_seed_update' '/dev/stdout' 2>&1 | tee -a "$setup_log"
+		update_docker_containers 'netinstall' '' 'docker_seed_update' '/dev/stdout' 2>&1 | tee -a "$setup_log"
         # Use pipe exit status of 'update_docker_containers' for return code
-	return ${PIPESTATUS[0]}
+		return ${PIPESTATUS[0]}
+	fi
 }
 
 elasticfleet_pillar() {