Wazuh server cluster

Master

The master node is responsible for:

• Receiving and managing agent registration requests.

• Creating shared configuration groups.

• Updating custom rules, decoders and CDB lists.

• Synchronizing all this information to the workers nodes.

All the outlined responsibilities of the master node listed above is called Integrity. It is synchronized from the master to the workers even if the workers have a more recent modification time or a higher size.

Master nodes can also receive and process events from agents the same way a worker would do.

Worker

A worker node is responsible for:

• Redirecting agent registration requests to the master node.

• Receiving updates from the master node.

• Receiving and processing events from agents.

• Sending last keep alive messages from agents and remoted group assignments to the master node.

If any integrity file is modified in a worker node, its content will be replaced with the version on the master node.

Keep alive thread

Worker nodes in the Wazuh server cluster periodically send keep-alive messages to the master node. The master node records the timestamp of the last keep-alive received and monitors the expected interval for each worker. If a worker node fails to send a keep-alive within the defined time limit, the master marks it as disconnected and closes the connection. Once the worker node detects the disconnection, it automatically attempts to reconnect to the master node.

This mechanism helps maintain cluster stability by automatically removing nodes that are unreachable or experiencing network issues.

Integrity thread

The integrity thread is responsible for synchronizing integrity data between the Wazuh server master node and all worker nodes. The synchronization process is initiated by the worker node and occurs in the following stages:

1. The worker node requests permission from the master node to start the synchronization process. Permission is granted only after any ongoing synchronization process has completed. This prevents overlap between concurrent synchronization operations. Synchronization processes that take too long are considered locked due to errors. When a process is marked as locked, new integrity synchronization requests can be approved.

   By default, the maximum time allowed for a synchronization process is 1000 seconds. This value can be modified with the max_locked_integrity_time variable in the cluster.json file.

2. The worker node sends the master node a JSON file containing the following information:

   • Path

   • Modification time

   • Blake2b checksum

   • Information showing whether the file is a merged file or not. If it is a merged file, it shows the following information:

     • The merge type

     • The filename

3. The master node compares the received checksums with its own and creates three different file groups:

   • Missing: Files that are present in the master node but are missing in the worker node. They must be created in the worker.

   • Extra: Files that are present in the worker node but missing in the master. They must be removed from the worker node.

   • Shared: Files that are present in both master and worker but have a different checksum. They must be updated on the worker node with the version from the master node.

   The master node then creates a ZIP package containing a JSON file with the required integrity information and any files the worker node needs to update. The maximum size of this package is defined by the max_zip_size variable in the cluster.json file. If the package exceeds this limit, the remaining files are synchronized during the next integrity synchronization cycle.

4. Once the worker node receives the package, it updates the necessary files. If there is no data to synchronize or there has been an error reading data from the worker, the worker is always notified about it. Also, if a timeout error occurs while the worker node is waiting to receive the zip, the master node will cancel the current task and reduce the zip size limit. The limit will gradually increase again if no new timeout errors occur.

Agent info thread

The agent info thread synchronizes the last keep-alive messages and operating system information from worker nodes with the Wazuh server master node. The synchronization process is initiated by the worker node and occurs in the following stages:

1. The worker node requests permission from the master node before starting synchronization. This prevents multiple synchronization processes from running at the same time.

2. The worker node queries the local wazuh-db service to retrieve information about agents marked as not synchronized.

3. The worker node sends the master node a JSON string containing the information retrieved from wazuh-db.

4. The master node sends the received information to its local wazuh-db service, where it is updated.

Agent groups send thread

The agent groups send thread synchronizes Wazuh agent group assignments (agent-groups) from the master node to all worker nodes in the Wazuh server cluster. Its purpose is to ensure that agent-group information stored on the master node is replicated in the databases of all worker nodes. The communication is initiated by the master node and follows these stages:

1. When there is new agent-groups information, the master node sends a JSON string with it to each worker node. This is done only once per worker node.

2. The worker nodes send the received information to their local wazuh-db service, where it is updated.

3. The worker nodes compare the checksum of their database with the checksum of the master node.

4. If the checksum has been different for 10 consecutive times, the worker node notifies the master node.

5. When notified, the master node sends all the agent-groups information to the worker node.

6. The worker node overwrites its database with the agent-groups information it has received from the master node.

Local agent-groups thread

The local agent-groups thread runs exclusively on the Wazuh server master node. It periodically queries the local wazuh-db service for any new or updated agent-group information since the last execution. The process doesn't repeat until all updated agent-group data has been successfully distributed to every worker node in the cluster.

Local integrity thread

The local integrity thread runs exclusively on the Wazuh server master node. It periodically reads all integrity files and calculates their checksums. Checksum calculation is slow and reduces performance in clusters with many worker nodes, since each worker node requires its own checksum. To optimize performance, this thread calculates the required integrity checksums once and stores them in a global variable that is periodically updated.

Sendsync thread

The sendsync thread is a Wazuh server cluster task executed exclusively on the master node. Although not shown in the workflow diagram, it enables the redirection of messages received from worker nodes to the appropriate master node services.

This mechanism allows agent registration requests to be processed even when agents are configured to connect to a worker node. In such cases, the sendsync thread receives the registration request from the worker node and forwards it to the master node's Authd service.

Distributed API thread

This thread isn't shown in the schema either. It runs in both master and worker nodes since it is independent of the node type. It is used to receive API requests and forward them to the most suitable node to process the request. The operation of this thread will be explained in later sections.

Protocol definition

The communication protocol used for all communications (both cluster and API) is defined in wazuh.core.cluster.common.Handler. Each message in the protocol has the following structure:

The protocol message has two parts:

1. The header: This is usually 22 bytes long. The header has four subparts:

   • Counter: It specifies the message ID. This is randomly generated for every new sent request. It is very useful when receiving a response, so it indicates which sent request it is replying to.

   • Payload length: It specifies the amount of data contained in the message payload. It is used to know how much data to expect to receive.

   • Command: It specifies the protocol message. This string will always be 11 characters long. If the command is not 11 characters long, a padding of - is added until the string reaches the expected length. All available commands in the protocol are shown below.

   • Flag message divided: This specifies whether the message has been divided depending on if the initial payload length was more than 5242880 bytes or not. The flag value can be d if the message is a divided one, or nothing if the message is the end of a divided message or a single message.

2. The payload: This will be 5242880 bytes long at maximum.

Wazuh cluster protocol

This communication protocol is used by all Wazuh server cluster nodes to synchronize information required to receive reports from the Wazuh agents. All communications are made through TCP. These commands are defined in wazuh.core.cluster.master.MasterHandler.process_request and in wazuh.core.cluster.worker.WorkerHandler.process_request.

Local protocol

This communication protocol is used by the Wazuh server API to forward requests to other Wazuh server cluster nodes. All communications are made using a Unix socket since the communication is local (from the process running the API to the process running the cluster). These commands are defined in wazuh.core.cluster.local_server.LocalServerHandler.process_request, wazuh.core.cluster.local_server.LocalServerHandlerMaster.process_request and wazuh.core.cluster.local_server.LocalServerHandlerWorker.process_request.

Common messages

All cluster communication protocols are built on a shared abstract base. This base layer defines the core messages used to manage connections, exchange keep-alive signals, and perform other essential cluster operations. These message-handling commands are implemented in the following modules: wazuh.core.cluster.common.Handler.process_request, wazuh.core.cluster.server.AbstractServerHandler.process_request and wazuh.core.cluster.client.AbstractClient.process_request.

Asynchronous tasks

A key factor contributing to the performance of the Wazuh server cluster is the use of asynchronous tasks. Similar to threads in Python, asynchronous tasks run concurrently with the main process and other tasks, but they are significantly more lightweight and faster to create. Their efficiency comes from leveraging the latency of I/O operations. This means that while one task is waiting for data to be transmitted or received through a socket, another can continue executing.

Each of the threads described in the Workflow section are implemented as asynchronous tasks. These tasks are started in wazuh.core.cluster.client.AbstractClientManager.start, wazuh.core.cluster.server.AbstractServer.start, and wazuh.core.cluster.local_server.LocalServer.start and they are all implemented using infinite loops.

In addition to the tasks already described, additional tasks are created whenever an incoming request requires more complex processing. These tasks are instantiated specifically to handle the request and are terminated once the corresponding response has been sent. This architectural approach ensures that the server is not blocked by a single request and can continue serving others efficiently.

One such task, implemented as a class, is responsible for receiving and processing files from a peer. An instance of this class is created at the start of a synchronization process and is destroyed once the process is complete. Internally, it launches an asynchronous task that remains active until all files required for synchronization have been successfully received. This asynchronous task has a callback that checks if there was an error during the synchronization process.

Multiprocessing

Asynchronous tasks are suitable for optimizing workloads and reducing I/O wait times but are unsuitable for CPU-intensive tasks because of Python's execution model. Python allows a single thread to take control of the Python interpreter through the Global Interpreter Lock (GIL). Therefore, asynchronous tasks run concurrently and not in parallel. Following the analogy of the previous section, it is as if there is effectively only one chef who has to do all the tasks. The chef can only do one at a time so if one task requires all his/her attention, the other ones are delayed.

The master node in the Wazuh server cluster executes a heavy workload, especially in large environments. Although the tasks are asynchronous, they have sections that require high CPU usage, such as calculating the hash of the files to be synchronized with the Local integrity thread. To avoid other tasks from waiting for the Python interpreter to complete the CPU-bound parts, multiprocessing is used. Using the same analogy again, multiprocessing would be equivalent to having more chefs working in the same kitchen.

Multiprocessing is implemented in the cluster process of both the Wazuh server master node and the worker nodes, and the concurrent.futures.ProcessPoolExecutor is used for this purpose. Cluster tasks can use any available process in the pool to delegate and run in parallel, the parts of their logic that are CPU intensive. With this, it is possible to take advantage of more CPU cores and increase the overall performance of the cluster process. When combined with asyncio, best results are obtained.

Child processes are created when the parent wazuh-clusterd starts. They stay in the process pool waiting for new jobs to be assigned to them. There are two child processes by default within the master node pool. This value can be changed in the process_pool_size variable in the cluster.json file. The worker nodes, on the other hand, create a single child process and this number is not modifiable. The tasks that use multiprocessing in the cluster are the following.

Worker nodes

• Integrity thread: This is the only task in the Wazuh server worker nodes that uses multiprocessing. It carries out the following CPU intensive actions:

  • Hash calculation: This calculates the hash of all the files to be synchronized every time Integrity check is started.

  • Unzip files: This extracts files and can take too long when the zip is large.

  • Process master files: This processes and moves all the files that were received from the Wazuh server master node to the appropriate destination.

Below is an example diagram of how the process pool is used in the Wazuh server master node: