Skip to content

Load standard and promoted test/custom content into all cluster nodes - #1399

Open
adrianvponce wants to merge 9 commits into
5.0.0from
enhancement/1376-load-standard-space-into-all-nodes
Open

Load standard and promoted test/custom content into all cluster nodes#1399
adrianvponce wants to merge 9 commits into
5.0.0from
enhancement/1376-load-standard-space-into-all-nodes

Conversation

@adrianvponce

@adrianvponce adrianvponce commented Jul 20, 2026

Copy link
Copy Markdown
Contributor

Description

In a Wazuh cluster, Engine content was loaded into the elected cluster-manager's Engine only. Because each node runs its own Engine over a node-local Unix domain socket, engine-backed operations
(e.g. POST /_plugins/_content_manager/logtest) only returned correct results when the request happened to be served by the manager.

This affects two independent content flows, and this PR distributes the per-node Engine load for both — while keeping the operations that must stay centralized on a single node. The content itself already lives in the cluster-wide wazuh-threatintel-* indices; only the node-local Engine load was missing on the other nodes.

1. STANDARD space (after a catalog sync)

STANDARD content is refreshed by the catalog sync, which runs manager-only. After a sync, every node's local Engine now converges to the new STANDARD content. The external catalog fetch stays manager-only (a single external pull); only the Engine load becomes distributed.

2. TEST / CUSTOM spaces (after a promotion)

Promoted TEST/CUSTOM content was deployed only to the node that served the POST /_plugins/_content_manager/promote request. TEST and CUSTOM are shared, cluster-persisted, singleton spaces, so their Engine representation should be consistent on every node. After a promote, every node's local Engine now loads the consolidated content. The promote's validation gate and consolidation stay on the coordinating node; the other nodes self-load from the replicated indices afterwards (no re-validation).

Shared mechanism: node-local load, cluster-wide trigger

engine.promote is a node-local Unix socket and cannot be made a cluster-wide call, so the load stays per-node and is driven by a cluster-wide signal:

  • A shared EngineContentLoader tracks the aggregate space.hash.sha256 of {standard, test, custom} and loads a space into the local Engine only when its hash differs from what this node last loaded (hash-gated, single-flighted, in-memory).
  • A per-node ClusterStateListener (registered on every node, outside the manager-only gate) drives the loader on cluster-state updates — this gives convergence for late/rejoining/new nodes and self-healing retry (a load that failed because the Engine was not ready is retried on a later event, no polling).
  • A ReloadEngineContentAction nodes-broadcast provides prompt propagation to already-running nodes: it is fired after a catalog sync (STANDARD) and after a promote (TEST/CUSTOM), and each node runs the same hash-gated loader.

Closes # #1376

Proposed Changes

All changes within plugins/content-manager/.

Shared loader + broadcast

  • New cti/catalog/service/EngineContentLoader.java — node-local, hash-gated loader over the shared spaces {standard, test, custom}; single-flighted, records a space's hash only on 200 OK so failed loads retry.
  • New action/ReloadEngineContentAction (+ ReloadEngineContentRequest, ReloadEngineContentNodeResponse, ReloadEngineContentResponse) and transport/TransportReloadEngineContentAction — a TransportNodesAction that runs reloadIfChanged() on every node.
  • utils/Constants.java — space-parameterized Engine-load log messages.

STANDARD path (catalog sync)

  • ContentManagerPlugin — instantiates the shared loader; registers the ClusterStateListener on all nodes; performs an immediate load attempt on startup; registers the broadcast action.
  • ConsumerRulesetService — fires the ReloadEngineContentAction broadcast after a content-changing sync (replacing its former direct EngineService dependency).
  • CatalogSyncJob — consequential one-line update to the ConsumerRulesetService construction (dropped argument).

TEST / CUSTOM path (promotion)

  • transport/TransportPostPromoteAction — after consolidation recomputes the target space's hash, fires the same ReloadEngineContentAction broadcast so every node loads the promoted TEST/CUSTOM content.

Tests

  • New EngineContentLoaderTests (per-space hash-gated load, single-flight, non-OK retry, multi-space coverage) and TransportReloadEngineContentActionTests (node delegation, response aggregation, wire round-trip).
  • Updated ConsumerRulesetServiceTests, ContentManagerPluginTests, SpaceInitializationIT.

Results and Evidence

See comments:

Artifacts Affected

wazuh-indexer-content-manager plugin (JAR)

Configuration Changes

N/A

Documentation Updates

N/A

Tests Introduced

New unit tests

  • EngineContentLoaderTests — the node-local loader: hash-gated load per space, single-flight collapse, loadedHash advanced only on 200 (non-OK is retried), null-policy / missing-hash skips, and multi-space coverage over {standard, test, custom} (all three load, only a changed space reloads, a space loads independently).
  • TransportReloadEngineContentActionTests — the broadcast nodes-action: nodeOperation delegates to the loader and reports the local node, newResponse aggregates under the cluster name, and per-node response wire round-trip.
    Existing tests updated
    • ConsumerRulesetServiceTests, CatalogSyncJobTests, TransportTriggerUpdateActionTests, ContentManagerPluginTests , SpaceInitializationIT.

Review Checklist

  • Code changes reviewed
  • Relevant evidence provided
  • Tests cover the new functionality
  • Configuration changes documented
  • Developer documentation reflects the changes
  • Meets requirements and/or definition of done
  • No unresolved dependencies with other issues
  • PR is linked to the relevant issue(s)
  • Correct labels applied (e.g., no-changelog)
  • ...

@adrianvponce adrianvponce self-assigned this Jul 20, 2026
@adrianvponce
adrianvponce force-pushed the enhancement/1376-load-standard-space-into-all-nodes branch from 2ae4a5a to caa3ad5 Compare July 20, 2026 20:17
@adrianvponce
adrianvponce force-pushed the enhancement/1376-load-standard-space-into-all-nodes branch from caa3ad5 to fd0a9df Compare July 21, 2026 13:06
@adrianvponce
adrianvponce marked this pull request as ready for review July 21, 2026 18:45
@adrianvponce
adrianvponce requested a review from a team as a code owner July 21, 2026 18:45
@adrianvponce adrianvponce linked an issue Jul 21, 2026 that may be closed by this pull request
5 tasks
@adrianvponce

adrianvponce commented Jul 22, 2026

Copy link
Copy Markdown
Contributor Author

Validation - "Load the standard space into the engine after a catalog sync"

Scope: per the issue, STANDARD-space content must be loaded into every node's Engine after a catalog sync. This run validates that on the live cluster with the patched (broadcast) build deployed on both nodes.

Environment: live 2-node cluster — node-1 = 192.168.121.9 (non-master), node-2 = 192.168.121.149 (cluster-manager), real Wazuh Engine on each.

Trigger: changed the subscription plan to Pro on the manager and ran a content update. The ruleset consumer synced and updated STANDARD content (the IoC/CVE consumers are unrelated to the standard-space engine load).

# change the subscription to Pro

PUT /_cluster/settings
{
 "persistent": {
   "plugins.content_manager.wazuh_uid": "<wazuh_uid>"
 }
}


POST /_plugins/_content_manager/subscription
{
 "access_token": "<access_token>"
}

# trigger the sync
POST /_plugins/_content_manager/update

Check Indexer logs:

[2026-07-27T19:43:03,970][INFO ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for standard space completed successfully.
[2026-07-27T19:43:04,079][INFO ][c.w.c.c.c.s.EngineContentLoader] [node-2] Engine load for standard space completed successfully.

@adrianvponce

adrianvponce commented Jul 22, 2026

Copy link
Copy Markdown
Contributor Author

Validation — "Load promoted test and custom spaces into all nodes"

# Create an integration

POST /_plugins/_content_manager/integrations
{
    "resource": {
        "category": "other",
        "description": "This is a test integration.",
        "documentation": "test1234",
        "references": [
            "https://wazuh.com"
        ],
        "title": "findings-enrichment-test-2",
        "metadata": {
          "author": "Wazuh Inc.",
          "title": "test"
        }
    }
}

{
  "message": "78a04887-6d9b-42d0-a346-2906223d5b25",
  "status": 201
}

# Create a decoder to be used as root_decoder

POST /_plugins/_content_manager/decoders
{
  "integration": "78a04887-6d9b-42d0-a346-2906223d5b25",
  "resource": {
    "enabled": true,
    "metadata": {
      "compatibility": "All wazuh events.",
      "description": """Base decoder to process Wazuh message format, parses location part and enriches the events that comes from a Wazuh agent with the host information.
""",
      "module": "wazuh",
      "references": [
        "https://documentation.wazuh.com/"
      ],
      "title": "Wazuh message decoder",
      "versions": [
        "Wazuh 5.*"
      ]
    },
    "name": "decoder/core-wazuh-message/2",
    "normalize": [
      {
        "map": [
          {
            "@timestamp": "get_date()"
          }
        ]
      }
    ]
  }
}


{
  "message": "92785e1e-8ace-475d-b705-a8c319bf645f",
  "status": 201
}


GET wazuh-threatintel-policies/_search?size=1
{
 "query": {
   "match": {
     "space.name": "draft"
   }
 }
}

# Update the draft policy

PUT /_plugins/_content_manager/policy/draft
{
  "resource": {
    "id": "4a2bbfac-1235-37ba-ac04-0cefe3ecace0",
    "metadata": {
      "title": "Custom space",
      "author": "Custom",
      "date": "2026-07-27T18:52:12Z",
      "modified": "2026-07-27T18:52:12Z",
      "description": "Custom space",
      "references": [],
      "documentation": "",
      "compatibility": []
    },
    "root_decoder": "92785e1e-8ace-475d-b705-a8c319bf645f",
    "integrations": [
      "78a04887-6d9b-42d0-a346-2906223d5b25"
    ],
    "filters": [],
    "enrichments": [],
    "enabled": true,
    "index_unclassified_events": false,
    "index_discarded_events": false
  }
}

{
  "message": "4acfc4c2-7f97-3c9a-b15d-23a2114b3e8d",
  "status": 200
}


GET _plugins/_content_manager/promote?space=test


# Promote to test

POST _plugins/_content_manager/promote
{
  "space": "draft",
  "changes": {
    "kvdbs": [],
    "decoders": [
      {
        "operation": "add",
        "id": "92785e1e-8ace-475d-b705-a8c319bf645f"
      }
    ],
    "rules": [],
    "filters": [],
    "integrations": [
      {
        "operation": "add",
        "id": "78a04887-6d9b-42d0-a346-2906223d5b25"
      }
    ],
    "policy": [
      {
        "operation": "update",
        "id": "4a2bbfac-1235-37ba-ac04-0cefe3ecace0"
      }
    ]
  }
}

{
  "message": "Promotion completed successfully.",
  "status": 200
}

Indexer logs:

[2026-07-27T19:10:31,565][INFO ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space completed successfully.
[2026-07-27T19:10:31,560][INFO ][c.w.c.c.c.s.EngineContentLoader] [node-2] Engine load for test space completed successfully.
# Promote to custom

POST _plugins/_content_manager/promote
{
  "space": "test",
  "changes": {
    "kvdbs": [],
    "decoders": [
      {
        "operation": "add",
        "id": "92785e1e-8ace-475d-b705-a8c319bf645f"
      }
    ],
    "rules": [],
    "filters": [],
    "integrations": [
      {
        "operation": "add",
        "id": "78a04887-6d9b-42d0-a346-2906223d5b25"
      }
    ],
    "policy": [
      {
        "operation": "update",
        "id": "4a2bbfac-1235-37ba-ac04-0cefe3ecace0"
      }
    ]
  }
}

{
  "message": "Promotion completed successfully.",
  "status": 200
}

Indexer logs:

[2026-07-27T19:11:49,955][INFO ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space completed successfully.
[2026-07-27T19:11:49,935][INFO ][c.w.c.c.c.s.EngineContentLoader] [node-2] Engine load for custom space completed successfully.

@adrianvponce adrianvponce changed the title Load standard space into all cluster nodes Load standard and promoted test/custom content into all cluster nodes Jul 22, 2026
@f-galland
f-galland self-requested a review July 27, 2026 09:15
@f-galland

f-galland commented Jul 27, 2026

Copy link
Copy Markdown
Member

Validation

Content tests pass 🟢

Details

$ ./run-validation.sh --host 192.168.56.100 --user admin --password admin
Target: https://192.168.56.100:9200/_plugins/_content_manager   user=admin   profile=admin

ACCESS_TOKEN not set — subscription / content-update steps will be skipped

» waiting for ≥3 .wazuh-cti-consumers to be ready (idle + local_offset>0, timeout 800s)
   … 1/2 ready (need ≥3, 0s elapsed)
   … 1/2 ready (need ≥3, 10s elapsed)
   … 1/2 ready (need ≥3, 20s elapsed)
   … 2/3 ready (need ≥3, 30s elapsed)
   … 2/3 ready (need ≥3, 40s elapsed)
   … 2/3 ready (need ≥3, 50s elapsed)
   … 2/3 ready (need ≥3, 60s elapsed)
   … 2/3 ready (need ≥3, 70s elapsed)
   … 2/3 ready (need ≥3, 80s elapsed)
   … 2/3 ready (need ≥3, 90s elapsed)
  PASS   consumers_idle         [3/3 ready, min 3]
  PASS   create_integration     [HTTP 201, want 201]
         INTEGRATION_ID = bd15b1b5-2077-4e64-883e-861b90a110fd
  PASS   get_integration        [HTTP 200, want 200] (assert ok)
  PASS   create_decoder         [HTTP 201, want 201]
         DECODER_ID = 8cd200f5-5b77-4a66-87dc-da6a18391ed0
  PASS   get_decoder            [HTTP 200, want 200] (assert ok)
  PASS   create_rule            [HTTP 201, want 201]
         RULE_ID = 3342983c-6956-4721-9591-fd96399bb778
  PASS   get_rule               [HTTP 200, want 200] (assert ok)
  PASS   create_kvdb            [HTTP 201, want 201]
         KVDB_ID = acd2c160-41b1-445d-9d29-4faced0c20b6
  PASS   get_kvdb               [HTTP 200, want 200] (assert ok)
  PASS   create_filter          [HTTP 201, want 201]
         FILTER_ID = 237bf33f-cea0-4752-9089-003711e837de
  PASS   get_filter             [HTTP 200, want 200] (assert ok)
  PASS   update_policy          [HTTP 200, want 200]
  PASS   get_policy             [HTTP 200, want 200] (assert ok)
  PASS   promote_preview_draft  [HTTP 200, want 200]
  PASS   promote_draft          [HTTP 200, want 200]
  PASS   promote_draft_check    [HTTP 200, want 200] (assert ok)
  PASS   promote_preview_test   [HTTP 200, want 200]
  PASS   promote_test           [HTTP 200, want 200]
  PASS   promote_test_check     [HTTP 200, want 200] (assert ok)
  PASS   update_integration     [HTTP 200, want 200]
  PASS   get_updated_integration [HTTP 200, want 200] (assert ok)
  PASS   update_decoder         [HTTP 200, want 200]
  PASS   get_updated_decoder    [HTTP 200, want 200] (assert ok)
  PASS   update_rule            [HTTP 200, want 200]
  PASS   get_updated_rule       [HTTP 200, want 200] (assert ok)
  PASS   update_kvdb            [HTTP 200, want 200]
  PASS   get_updated_kvdb       [HTTP 200, want 200] (assert ok)
  PASS   update_filter          [HTTP 200, want 200]
  PASS   get_updated_filter     [HTTP 200, want 200] (assert ok)
  PASS   promote_preview_draft2 [HTTP 200, want 200]
  PASS   promote_draft2         [HTTP 200, want 200]
  PASS   promote_draft2_check   [HTTP 200, want 200] (assert ok)
  PASS   promote_preview_test2  [HTTP 200, want 200]
  PASS   promote_test2          [HTTP 200, want 200]
  PASS   promote_test2_check    [HTTP 200, want 200] (assert ok)
  PASS   logtest                [HTTP 200, want 200]
  PASS   logtest_result         [response assert ok]
  PASS   reset_draft            [HTTP 200, want 200]
  PASS   reset_draft_check      [HTTP 200, want 200] (assert ok)
» ACCESS_TOKEN not set — skipping subscription / content-update steps.
  (export ACCESS_TOKEN=<token> to exercise them.)
» waiting for ≥3 .wazuh-cti-consumers to be ready (idle + local_offset>0, timeout 800s)
  PASS   consumers_idle         [3/3 ready, min 3]
  PASS   version_check          [HTTP 200, want 200]
  PASS   version_check_result   [response assert ok]

» DELETE coverage: create a keeper integration + decoder for the policy root_decoder
  PASS   dc_create_keeper_integration [HTTP 201, want 201]
         INTEGRATION_ID = 9dbb045d-8078-4534-9a9c-30a44c82a1f3
  PASS   dc_create_keeper_decoder [HTTP 201, want 201]
         KEEPER_DECODER_ID = efdd4ad0-db03-4d9f-a29c-4c060ef4914c
  PASS   dc_update_policy       [HTTP 200, want 200]

» DELETE coverage: create the delete-target set
  PASS   dc_create_integration  [HTTP 201, want 201]
         INTEGRATION_ID = a3952554-bd15-4b60-8659-357ccd41187d
  PASS   dc_create_decoder      [HTTP 201, want 201]
         DECODER_ID = c4c327f6-b13f-4c47-9a52-c6841df99909
  PASS   dc_create_rule         [HTTP 201, want 201]
         RULE_ID = 9d4d980b-303b-451d-b5db-d540d542de8c
  PASS   dc_create_kvdb         [HTTP 201, want 201]
         KVDB_ID = fce8510a-87cb-4426-aaab-cff3bd70ea8e
  PASS   dc_create_filter       [HTTP 201, want 201]
         FILTER_ID = 3d02d3bb-557f-41b1-ad3e-7096c673444e
  PASS   delete_decoder         [HTTP 200, want 200]
  PASS   delete_decoder_check   [HTTP 200, want 200] (assert ok)
  PASS   delete_rule            [HTTP 200, want 200]
  PASS   delete_rule_check      [HTTP 200, want 200] (assert ok)
  PASS   delete_kvdb            [HTTP 200, want 200]
  PASS   delete_kvdb_check      [HTTP 200, want 200] (assert ok)
  PASS   delete_filter          [HTTP 200, want 200]
  PASS   delete_filter_check    [HTTP 200, want 200] (assert ok)

» DELETE coverage: promote (pushes the delete-target integration up)
  PASS   dc_promote_preview_draft [HTTP 200, want 200]
  PASS   dc_promote_draft       [HTTP 200, want 200]
  PASS   dc_promote_preview_test [HTTP 200, want 200]
  PASS   dc_promote_test        [HTTP 200, want 200]
  PASS   delete_integration     [HTTP 200, want 200]
  PASS   delete_integration_check [HTTP 200, want 200] (assert ok)

» DELETE coverage: promote the integration deletion up to custom
  PASS   dc_promote_preview_draft2 [HTTP 200, want 200]
  PASS   dc_promote_draft2      [HTTP 200, want 200]
  PASS   dc_promote_preview_test2 [HTTP 200, want 200]
  PASS   dc_promote_test2       [HTTP 200, want 200]
  PASS   delete_integration_propagated [HTTP 200, want 200] (assert ok)

──────────────────────────────────────────────
  PASS: 69   DENIED: 0   SKIP: 0   FAIL: 0
──────────────────────────────────────────────

Content syncs without error 🔴

Note

The following error logs show up in both nodes of a 2-node cluster

Details

[2026-07-27T09:33:12,962][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_b187a4': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:12,976][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_62e88f': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,231][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_dfca60': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,249][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_5f4b4a': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,360][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_861929': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,371][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_6df5ee': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,440][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_f59de2': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,452][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_a7af61': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,937][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_d31f81': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:13,951][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_6ae193': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:14,493][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_4a096c': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:14,503][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_56279b': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:14,739][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_54fa23': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:33:14,756][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_468dd9': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:36:11,925][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_9f8577': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:36:11,933][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_e0b221': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:36:11,982][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for test space returned status [400]: Failed to import namespace 'logtest_1a721b': Policy JSON must have a 'root_decoder' field
[2026-07-27T09:36:11,989][WARN ][c.w.c.c.c.s.EngineContentLoader] [node-1] Engine load for custom space returned status [400]: Failed to import namespace 'logtest_67eb46': Policy JSON must have a 'root_decoder' field

These logs do not show up in a standard 5.0.0 install

@f-galland f-galland left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Error logs were found here

We are also using the generic threadpool, which we are actively avoiding in favor of Client's async methods.

}
try {
this.threadPool
.generic()

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is a limited amount of generic threads, this could block execution of other relevant tasks.

@adrianvponce
adrianvponce requested a review from f-galland July 27, 2026 19:52
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Wazuh content is only loaded on the cluster-manager Engine

2 participants