359 lines
12 KiB
Perl
359 lines
12 KiB
Perl
package PandoraFMS::PredictionServer;
|
|
########################################################################
|
|
# Pandora FMS Prediction Server.
|
|
# Pandora FMS. the Flexible Monitoring System. http://www.pandorafms.org
|
|
########################################################################
|
|
# Copyright (c) 2005-2021 Artica Soluciones Tecnologicas S.L
|
|
#
|
|
# This program is free software; you can redistribute it and/or
|
|
# modify it under the terms of the GNU Lesser General Public License
|
|
# as published by the Free Software Foundation; version 2
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software
|
|
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
|
|
##########################################################################
|
|
|
|
use strict;
|
|
use warnings;
|
|
|
|
use threads;
|
|
use threads::shared;
|
|
use Thread::Semaphore;
|
|
|
|
use IO::Socket::INET;
|
|
use Net::Ping;
|
|
use POSIX qw(strftime);
|
|
|
|
# Default lib dir for RPM and DEB packages
|
|
use lib '/usr/lib/perl5';
|
|
|
|
use PandoraFMS::Tools;
|
|
use PandoraFMS::DB;
|
|
use PandoraFMS::Core;
|
|
use PandoraFMS::ProducerConsumerServer;
|
|
|
|
#For debug
|
|
#use Data::Dumper;
|
|
|
|
# Inherits from PandoraFMS::ProducerConsumerServer
|
|
our @ISA = qw(PandoraFMS::ProducerConsumerServer);
|
|
|
|
# Global variables
|
|
my @TaskQueue :shared;
|
|
my %PendingTasks :shared;
|
|
my $Sem :shared;
|
|
my $TaskSem :shared;
|
|
|
|
########################################################################
|
|
# Prediction Server class constructor.
|
|
########################################################################
|
|
sub new ($$;$) {
|
|
my ($class, $config, $dbh) = @_;
|
|
|
|
return undef unless $config->{'predictionserver'} == 1;
|
|
|
|
# Initialize semaphores and queues
|
|
@TaskQueue = ();
|
|
%PendingTasks = ();
|
|
$Sem = Thread::Semaphore->new;
|
|
$TaskSem = Thread::Semaphore->new (0);
|
|
|
|
# Call the constructor of the parent class
|
|
my $self = $class->SUPER::new($config, PREDICTIONSERVER, \&PandoraFMS::PredictionServer::data_producer, \&PandoraFMS::PredictionServer::data_consumer, $dbh);
|
|
|
|
bless $self, $class;
|
|
|
|
return $self;
|
|
}
|
|
|
|
########################################################################
|
|
# Run.
|
|
########################################################################
|
|
sub run ($) {
|
|
my $self = shift;
|
|
my $pa_config = $self->getConfig ();
|
|
|
|
print_message ($pa_config, " [*] Starting " . $pa_config->{'rb_product_name'} . " Prediction Server.", 1);
|
|
$self->setNumThreads ($pa_config->{'prediction_threads'});
|
|
$self->SUPER::run (\@TaskQueue, \%PendingTasks, $Sem, $TaskSem);
|
|
}
|
|
|
|
########################################################################
|
|
# Data producer.
|
|
########################################################################
|
|
sub data_producer ($) {
|
|
my $self = shift;
|
|
my ($pa_config, $dbh) = ($self->getConfig (), $self->getDBH ());
|
|
|
|
my @tasks;
|
|
my @rows;
|
|
|
|
if (pandora_is_master($pa_config) == 0) {
|
|
@rows = get_db_rows ($dbh, 'SELECT tagente_modulo.id_agente_modulo,
|
|
tagente_modulo.flag, last_execution_try
|
|
FROM tagente, tagente_modulo, tagente_estado
|
|
WHERE server_name = ?
|
|
AND tagente_modulo.id_agente = tagente.id_agente
|
|
AND tagente.disabled = 0
|
|
AND tagente_modulo.prediction_module != 0
|
|
AND tagente_modulo.disabled = 0
|
|
AND tagente_estado.id_agente_modulo = tagente_modulo.id_agente_modulo
|
|
AND tagente_modulo.id_modulo = 5
|
|
AND (tagente_modulo.flag = 1
|
|
OR (tagente_estado.last_execution_try + tagente_estado.current_interval) < UNIX_TIMESTAMP())
|
|
ORDER BY last_execution_try ASC ', safe_input($pa_config->{'servername'}));
|
|
}
|
|
else {
|
|
# If is metaconsole server, will evaluate orphan modules also.
|
|
@rows = get_db_rows ($dbh, 'SELECT DISTINCT(tagente_modulo.id_agente_modulo),
|
|
tagente_modulo.flag, last_execution_try
|
|
FROM tagente, tagente_modulo, tagente_estado
|
|
WHERE ((server_name = ?)
|
|
OR (server_name = ANY(SELECT name
|
|
FROM tserver
|
|
WHERE status <> 1 AND server_type = ?))
|
|
OR ((server_name = 0 OR server_name IS NULL) AND 1=?)
|
|
)
|
|
AND tagente_modulo.id_agente = tagente.id_agente
|
|
AND tagente.disabled = 0
|
|
AND tagente_modulo.disabled = 0
|
|
AND tagente_modulo.prediction_module != 0
|
|
AND tagente_estado.id_agente_modulo = tagente_modulo.id_agente_modulo
|
|
AND tagente_modulo.id_modulo = 5
|
|
AND (tagente_modulo.flag = 1
|
|
OR (tagente_estado.last_execution_try + tagente_estado.current_interval) < UNIX_TIMESTAMP())
|
|
ORDER BY last_execution_try ASC',
|
|
safe_input($pa_config->{'servername'}),
|
|
PREDICTIONSERVER,
|
|
is_metaconsole($pa_config)
|
|
);
|
|
}
|
|
|
|
foreach my $row (@rows) {
|
|
|
|
# Reset forced execution flag
|
|
if ($row->{'flag'} == 1) {
|
|
db_do ($dbh, 'UPDATE tagente_modulo SET flag = 0 WHERE id_agente_modulo = ?', $row->{'id_agente_modulo'});
|
|
}
|
|
|
|
push (@tasks, $row->{'id_agente_modulo'});
|
|
}
|
|
|
|
return @tasks;
|
|
}
|
|
|
|
########################################################################
|
|
# Data consumer.
|
|
########################################################################
|
|
sub data_consumer ($$) {
|
|
my ($self, $task) = @_;
|
|
|
|
exec_prediction_module ($self->getConfig (), $task, $self->getServerID (), $self->getDBH ());
|
|
}
|
|
|
|
########################################################################
|
|
# Execute prediction module.
|
|
########################################################################
|
|
sub exec_prediction_module ($$$$) {
|
|
my ($pa_config, $id_am, $server_id, $dbh) = @_;
|
|
|
|
# Get a full hash for agent_module record reference ($agent_module)
|
|
my $agent_module = get_db_single_row ($dbh, 'SELECT *
|
|
FROM tagente_modulo
|
|
WHERE id_agente_modulo = ?', $id_am);
|
|
return unless defined $agent_module;
|
|
|
|
# Service modules
|
|
if ($agent_module->{'prediction_module'} == 2) {
|
|
|
|
if ($agent_module->{'custom_string_1'} eq 'SLA') {
|
|
logger ($pa_config, "Executing service module SLA " .
|
|
$agent_module->{'id_agente_modulo'} . " " .
|
|
$agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_service_module_sla', [$pa_config, $agent_module, $server_id, $dbh]);
|
|
}
|
|
elsif ($agent_module->{'custom_string_1'} eq 'SLA_Value') {
|
|
#Do none
|
|
}
|
|
else {
|
|
logger ($pa_config, "Executing service module " .
|
|
$agent_module->{'id_agente_modulo'} . " " .
|
|
$agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_service_module', [$pa_config, $agent_module, undef, $server_id, $dbh]);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
# Synthetic modules
|
|
if ($agent_module->{'prediction_module'} == 3) {
|
|
logger ($pa_config, "Executing synthetic module " . $agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_synthetic_module', [$pa_config, $agent_module, $server_id, $dbh]);
|
|
return;
|
|
}
|
|
|
|
# Netflow modules
|
|
if ($agent_module->{'prediction_module'} == 4) {
|
|
logger ($pa_config, "Executing netflow module " . $agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_netflow_module', [$pa_config, $agent_module, $server_id, $dbh]);
|
|
return;
|
|
}
|
|
|
|
# Cluster status module.
|
|
if ($agent_module->{'prediction_module'} == 5) {
|
|
logger ($pa_config, "Executing cluster status module " . $agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_cluster_status_module', [$pa_config, $agent_module, $server_id, $dbh]);
|
|
return;
|
|
}
|
|
|
|
# Cluster active-active module.
|
|
if ($agent_module->{'prediction_module'} == 6) {
|
|
logger ($pa_config, "Executing cluster active-active module " . $agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_cluster_aa_module', [$pa_config, $agent_module, $server_id, $dbh]);
|
|
return;
|
|
}
|
|
|
|
# Cluster active-passive module.
|
|
if ($agent_module->{'prediction_module'} == 7) {
|
|
logger ($pa_config, "Executing cluster active-passive module " . $agent_module->{'nombre'}, 10);
|
|
enterprise_hook ('exec_cluster_ap_module', [$pa_config, $agent_module, $server_id, $dbh]);
|
|
return;
|
|
}
|
|
|
|
# Get a full hash for target agent_module record reference ($target_module)
|
|
my $target_module = get_db_single_row ($dbh, 'SELECT * FROM tagente_modulo WHERE id_agente_modulo = ?', $agent_module->{'custom_integer_1'});
|
|
return unless defined $target_module;
|
|
|
|
# Prediction mode explanation
|
|
#
|
|
# 0 is for target type of generic_proc. It compares latest data with current data. Needs to get
|
|
# data on a "middle" interval, so if interval is 300, get data to compare with 150 before
|
|
# and 150 in the future. If current data is ABOVE or BELOW average +- typical_deviation
|
|
# this is a BAD value (0), if not is ok (1) and written in target module as is.
|
|
# more interval configured for this module, more "margin" has to compare data.
|
|
#
|
|
# 1 is for target type of generic_data. It get's data in the future, using the interval given in
|
|
# module. It gets average from current timestamp to INTERVAL in the future and gets average
|
|
# value. Typical deviation is not used here.
|
|
|
|
# 0 proc, 1 data
|
|
my $prediction_mode = ($agent_module->{'id_tipo_modulo'} == 2) ? 0 : 1;
|
|
|
|
# Initialize another global sub variables.
|
|
my $module_data = 0; # 0 data for default
|
|
|
|
# Get current timestamp
|
|
my $utimestamp = time ();
|
|
my $timestamp = strftime ("%Y-%m-%d %H:%M:%S", localtime($utimestamp));
|
|
|
|
# Get different data from each week one month ago (4 values)
|
|
# $agent_module->{'module_interval'} uses a margin of interval to get average data from the past
|
|
my @week_data;
|
|
my @week_utimestamp;
|
|
|
|
for (my $i=0; $i<4; $i++) {
|
|
$week_utimestamp[$i] = $utimestamp - (84600*7*($i+1));
|
|
# Adjust for proc prediction
|
|
if ($prediction_mode == 0) {
|
|
$week_utimestamp[$i] = $week_utimestamp[$i] - ($agent_module->{'module_interval'} / 2);
|
|
}
|
|
}
|
|
|
|
# Let's calculate statistical average using past data
|
|
# n = total of real data values
|
|
my ($n, $average, $temp1) = (0, 0, 0);
|
|
for (my $i=0; $i < 4; $i++) {
|
|
my ($first_data, $last_data, $average_interval);
|
|
my $sum_data = 0;
|
|
|
|
$temp1 = $week_utimestamp[$i] + $agent_module->{'module_interval'};
|
|
# Get data for week $i in the past
|
|
|
|
$average_interval = get_db_value ($dbh, 'SELECT AVG(datos)
|
|
FROM tagente_datos
|
|
WHERE id_agente_modulo = ?
|
|
AND utimestamp > ?
|
|
AND utimestamp < ?', $target_module->{'id_agente_modulo'}, $week_utimestamp[$i], $temp1);
|
|
|
|
# Need to get data outside interval because no data.
|
|
if (!(defined($average_interval)) || ($average_interval == 0)) {
|
|
$last_data = get_db_value ($dbh, 'SELECT datos
|
|
FROM tagente_datos
|
|
WHERE id_agente_modulo = ?
|
|
AND utimestamp > ?
|
|
LIMIT 1', $target_module->{'id_agente_modulo'}, $week_utimestamp[$i]);
|
|
next unless defined ($last_data);
|
|
$first_data = get_db_value ($dbh, 'SELECT datos
|
|
FROM tagente_datos
|
|
WHERE id_agente_modulo = ?
|
|
AND utimestamp < ?
|
|
LIMIT 1', $target_module->{'id_agente_modulo'}, $temp1);
|
|
next unless defined ($first_data);
|
|
$sum_data++ if ($last_data != 0);
|
|
$sum_data++ if ($first_data != 0);
|
|
$week_data[$i] = ($sum_data > 0) ? (($last_data + $first_data) / $sum_data) : 0;
|
|
}
|
|
else {
|
|
$week_data[$i] = $average_interval;
|
|
}
|
|
|
|
# It's possible that one of the week_data[i] values was not valid (NULL)
|
|
# so recheck it and relay on n=0 for "no data" values set to 0 in result
|
|
# Calculate total ammount of valida data for each data sample
|
|
if ((is_numeric($week_data[$i])) && ($week_data[$i] > 0)) {
|
|
$n++;
|
|
# Average SUM
|
|
$average = $average + $week_data[$i];
|
|
}
|
|
}
|
|
|
|
# Real average value
|
|
$average = ($n > 0) ? ($average / $n) : 0;
|
|
|
|
# (PROC) Compare with current data
|
|
if ($prediction_mode == 0) {
|
|
# Calculate typical deviation
|
|
my $typical_deviation = 0;
|
|
for (my $i=0; $i< $n; $i++) {
|
|
if ((is_numeric($week_data[$i])) && ($week_data[$i] > 0)) {
|
|
$typical_deviation = $typical_deviation + (($week_data[$i] - $average)**2);
|
|
}
|
|
}
|
|
$typical_deviation = ($n > 1) ? sqrt ($typical_deviation / ($n-1)) : 0;
|
|
|
|
my $current_value = get_db_value ($dbh, 'SELECT datos
|
|
FROM tagente_estado
|
|
WHERE id_agente_modulo = ?', $target_module->{'id_agente_modulo'});
|
|
if ( ($current_value > ($average - $typical_deviation)) && ($current_value < ($average + $typical_deviation)) ){
|
|
$module_data = 1; # OK !!
|
|
}
|
|
else {
|
|
$module_data = 0; # Out of predictions
|
|
}
|
|
}
|
|
else {
|
|
# Prediction based on data
|
|
$module_data = $average;
|
|
}
|
|
|
|
my %data = ("data" => $module_data);
|
|
pandora_process_module ($pa_config, \%data, '', $agent_module, '', $timestamp, $utimestamp, $server_id, $dbh);
|
|
|
|
my $agent_os_version = get_db_value ($dbh, 'SELECT os_version
|
|
FROM tagente
|
|
WHERE id_agente = ?', $agent_module->{'id_agente'});
|
|
|
|
if ($agent_os_version eq ''){
|
|
$agent_os_version = $pa_config->{'servername'}.'_Prediction';
|
|
}
|
|
|
|
pandora_update_agent ($pa_config, $timestamp, $agent_module->{'id_agente'}, undef, undef, -1, $dbh);
|
|
}
|
|
|
|
1;
|
|
__END__
|