Use broadcast system in agent-groups sync

[Selutario]

Description

It is requested to replace the current mechanism for obtaining and sending agent-groups data from master to the workers. Instead, it is intended to use the broadcast system that was developed here:

• Add broadcasting system for the cluster #13124

In the previous PR there are some useful examples for the implementation.

Checks

wazuh/wazuh

• Unit tests without failures. Updated and/or expanded if there are new functions/methods/outputs:
  • Cluster (framework/wazuh/core/cluster/tests/ & framework/wazuh/core/cluster/dapi/tests/)
  • Core (framework/wazuh/core/tests/)
  • SDK (framework/wazuh/tests/)
  • RBAC (framework/wazuh/rbac/tests/)
  • API (api/api/tests/)
• API tavern integration tests without failures. Updated and/or expanded if needed (api/test/integration/):
  • Affected tests
  • Affected RBAC (black and white) tests
• Review integration test mapping using the script (api/test/integration/mapping/integration_test_api_endpoints.json)
• Review of spec.yaml examples and schemas (api/api/spec/spec.yaml)
• Review exceptions remediation when any endpoint path changes or is removed (framework/wazuh/core/exception.py)
• Changelog (CHANGELOG.md)

wazuh/wazuh-documentation

• Migration from 3.X for changed endpoints (source/user-manual/api/equivalence.rst)
• Update RBAC reference with new/modified actions/resources/relationships (source/user-manual/api/rbac/reference.rst)

[Selutario]

Status update

Many of the variables and functions related to sending info to all agents have been removed:

wazuh/framework/wazuh/core/cluster/master.py

Lines 1070 to 1091 in 8dfd770

	def get_agent_groups_info(self, client):
	"""Check whether the updated group information is sent only once per worker.
	The variable with this information will not be updated until all the MasterHandlers send
	the information to their worker node.
	Parameters
	----------
	client : str
	String with the node name.
	Returns
	-------
	result : dict
	Updated data on agent-groups.
	"""
	result = {}
	if client in self.clients.keys() and client not in self.agent_groups_control_workers:
	result = self.agent_groups_control
	self.agent_groups_control_workers.add(client)
	return result

Also, this function will be called repeatedly instead of having an infinite loop as before. Therefore, the objects must be declared in a different place (probably in hello):

wazuh/framework/wazuh/core/cluster/master.py

Lines 711 to 722 in 8dfd770

	async def send_agent_groups_information(self):
	"""Function in charge of sending the group information to the worker node.
	Each time we get data it will be sent.
	A worker node cannot send two consecutive times the same group information.
	"""
	logger = self.task_loggers['Agent-groups send']
	wdb_conn = WazuhDBConnection()
	sync_object = c_common.SyncWazuhdb(manager=self, logger=logger, cmd=b'syn_g_m_w',
	data_retriever=wdb_conn.run_wdb_command,
	set_data_command='global set-agent-groups',
	set_payload={'mode': 'override', 'sync_status': 'synced'})

The development is not complete yet but everything worked fine in tests done during #13124.

To do

• Manual tests.
• Update unit tests.
• Run AIT
• Run QA system tests.

[Selutario]

Status update

The development is finished. As you can see below, the Local agent-groups task adds a method to the queue, which is concurrently executed by all workers right after it (Agent-groups send):

2022/06/17 12:53:01 INFO: [Master] [Local agent-groups] Starting.
2022/06/17 12:53:01 INFO: [Master] [Local agent-groups] Finished in 0.000s.
2022/06/17 12:53:01 INFO: [Worker worker1] [Agent-groups send] Starting.
2022/06/17 12:53:01 INFO: [Worker worker2] [Agent-groups send] Starting.
2022/06/17 12:53:01 INFO: [Worker worker1] [Agent-groups send] Finished in 0.004s. Updated 1 chunks.
2022/06/17 12:53:01 INFO: [Worker worker2] [Agent-groups send] Finished in 0.004s. Updated 1 chunks.

Everything worked fine in my manual tests. I updated unit tests and API integration tests are also passing. However, there is a problem unrelated to these changes in the system tests, which fail since the agents do not connect to the cluster in the test environments. Said error happens also in master branch.

To do

• Manual tests.
• Update unit tests.
• Run AIT
• Run QA system tests.