Merge branch 'ent-9170-sistema-de-procesamiento-paralelo-para-el-buffer-de-xmls' into 'develop'

Add support for smart queuing to the DataServer.

See merge request artica/pandorafms!5156

pandora_server/FreeBSD/pandora_server.conf.new

@@ -115,6 +115,10 @@ networkserver 1
 
 dataserver 1
 
+# Enable (1) or disable (0) the Data Server smart queue, which gives priority
+# to new data coming from agents at the expense of buffered XML files.
+dataserver_smart_queue 1
+
 # Activate (1) Pandora FMS Recon server
 
 reconserver 1

pandora_server/NetBSD/pandora_server.conf.new

@@ -110,6 +110,10 @@ networkserver 1
 
 dataserver 1
 
+# Enable (1) or disable (0) the Data Server smart queue, which gives priority
+# to new data coming from agents at the expense of buffered XML files.
+dataserver_smart_queue 1
+
 # Activate (1) Pandora FMS Recon server
 
 reconserver 1

pandora_server/conf/pandora_server.conf.new

@@ -145,6 +145,10 @@ networkserver 1
 
 dataserver 1
 
+# Enable (1) or disable (0) the Data Server smart queue, which gives priority
+# to new data coming from agents at the expense of buffered XML files.
+dataserver_smart_queue 1
+
 # Activate (1) Pandora FMS Discovery server
 
 discoveryserver 1

pandora_server/conf/pandora_server.conf.windows

@@ -119,6 +119,10 @@ networkserver 1
 
 dataserver 1
 
+# Enable (1) or disable (0) the Data Server smart queue, which gives priority
+# to new data coming from agents at the expense of buffered XML files.
+dataserver_smart_queue 1
+
 # Activate (1) Pandora FMS Recon server
 
 reconserver 1

pandora_server/lib/PandoraFMS/Config.pm

@@ -560,6 +560,8 @@ sub pandora_load_config {
 	$pa_config->{"tentacle_service_cmd"} = 'service tentacle_serverd'; # 7.0 761
 	$pa_config->{"tentacle_service_watchdog"} = 1; # 7.0 761
 
+	$pa_config->{"dataserver_smart_queue"} = 0; # 765.
+
 	# Check for UID0
 	if ($pa_config->{"quiet"} != 0){
 		if ($> == 0){
@@ -1326,6 +1328,9 @@ sub pandora_load_config {
 		elsif ($parametro =~ m/^ha_max_splitbrain_retries\s+([0-9]*)/i) {
 			$pa_config->{'ha_max_splitbrain_retries'} = clean_blank($1);
 		}
+		elsif ($parametro =~ m/^dataserver_smart_queue\s([0-1])/i) {
+			$pa_config->{'dataserver_smart_queue'} = clean_blank($1);
+		}
 		
 	} # end of loop for parameter #
 

pandora_server/lib/PandoraFMS/DataServer.pm

@@ -79,7 +79,13 @@ sub new ($$;$) {
 	$XMLinSem = Thread::Semaphore->new (1);
 	
 	# Call the constructor of the parent class
-	my $self = $class->SUPER::new($config, DATASERVER, \&PandoraFMS::DataServer::data_producer, \&PandoraFMS::DataServer::data_consumer, $dbh);
+	my $self;
+	if ($config->{'dataserver_smart_queue'} == 0) {
+		$self = $class->SUPER::new($config, DATASERVER, \&PandoraFMS::DataServer::data_producer, \&PandoraFMS::DataServer::data_consumer, $dbh);
+	} else {
+		logger($config, "Smart queue enabled for the Pandora FMS DataServer.", 3);
+		$self = $class->SUPER::new($config, DATASERVER, \&PandoraFMS::DataServer::data_producer_smart_queue, \&PandoraFMS::DataServer::data_consumer, $dbh);
+	}
 
 	# Load external .enc files for XML::Parser.
 	if ($config->{'enc_dir'} ne '') {
@@ -179,6 +185,53 @@ sub data_producer ($) {
 	return @tasks;
 }
 
+###############################################################################
+# Data producer with smart queuing.
+###############################################################################
+sub data_producer_smart_queue ($) {
+	my $self = shift;
+	my ($pa_config, $dbh) = ($self->getConfig (), $self->getDBH ());
+
+	my @tasks;
+	my @files;
+	my @sorted;
+
+	# Open the incoming directory
+	opendir (DIR, $pa_config->{'incomingdir'})
+		|| die "[FATAL] Cannot open Incoming data directory at " . $pa_config->{'incomingdir'} . ": $!";
+
+	# Do not read more than max_queue_files files
+	my $smart_queue = {};
+ 	while (my $file = readdir (DIR)) {
+ 		$file = Encode::decode( locale_fs => $file );
+
+		# Data files must have the extension .data
+		next if ($file !~ /^(.*)[\._]\d+\.data$/);
+		my $agent_name = $1;
+
+		# Queue a new file.
+		if (!defined($smart_queue->{$agent_name})) {
+			$smart_queue->{$agent_name} = $file;
+		}
+		# Or update a file in the queue.
+		else {
+			# Always work in LIFO mode.
+			if (-M $pa_config->{'incomingdir'} . '/' . $file < -M $pa_config->{'incomingdir'} . '/' . $smart_queue->{$agent_name}) {
+				$smart_queue->{$agent_name} = $file;
+			}
+		}
+	}
+	closedir(DIR);
+
+	# Do not process more than one XML from the same agent at the same time:
+	while (my ($agent_name, $file) = each(%{$smart_queue})) {
+		next if (agent_lock($pa_config, $dbh, $agent_name) == 0);
+		push (@tasks, $file);
+	}
+
+	return @tasks;
+}
+
 ###############################################################################
 # Data consumer.
 ###############################################################################