subagent-shell.in

#!@PERL@

use strict;
use warnings;
use Sys::Syslog qw(:standard :macros);
use Sys::Hostname;
use NetSNMP::agent (':all');
use NetSNMP::default_store (':all');
use NetSNMP::agent::default_store (':all');
use NetSNMP::ASN qw(:all);
use NetSNMP::OID (':all');
use SNMP;
use IPC::SysV qw(IPC_CREAT IPC_PRIVATE IPC_RMID S_IRWXU S_IRUSR S_IWUSR IPC_NOWAIT);
use Socket;
use POSIX qw(:sys_wait_h strftime locale_h);
use Time::HiRes qw( gettimeofday );
use Sys::Syslog qw(:standard :macros);
use File::Basename qw( basename fileparse );
use Getopt::Long;
use XML::Simple;
use Data::Dumper;

#disable warning read_config_store open
$ENV{SNMP_PERSISTENT_FILE}="/tmp/subagent-shell.$$";

our ($config, @functions_config, $reg_oid);
my  ($shm_id, $sem_id, $mQueue);
my  $running=1;
my  %agentMib = ( rootMIB => 'SUBAGENT-SHELL',
                  functionsCount => 'ssFunctionsCount',
                  functionsSuccessCount => 'ssFunctionsSuccessCount', 
                  functionsFailedCount => 'ssFunctionsFailedCount', 
                  timeStamp => 'ssTimeStamp', 
                  totalExecTime => 'ssTotalExecTime' );

my $opt = {
  daemon => 0,
  pid_file => '',
  base_dir => '/etc/snmp/subagent-shell/',
  mib_dir => '',
  agentx_socket => '',
  log_file => '',
  verbose => 0,
  errors => 0,
  help => 0,
};

Getopt::Long::Configure (qw(bundling pass_through));

# Get options.
Getopt::Long::GetOptions(
  'daemon|d' => \$opt->{daemon},
  'pid-file|p=s' => \$opt->{pid_file},
  'base-dir|b=s' => \$opt->{base_dir},
  'mib-dir|m=s' => \$opt->{mib_dir},
  'agentx-socket=s' => \$opt->{agentx_socket},
  'log-file|l=s' => \$opt->{log_file},
  'verbose|v+' => \$opt->{verbose},
  'exit-on-error|e+' => \$opt->{errors},
  'help|h' => \$opt->{help},
);

our ( $subAgentDir,$mibDir ); 

$subAgentDir = $opt->{base_dir}  unless defined($subAgentDir);
$subAgentDir = "${subAgentDir}/" unless $subAgentDir =~ /\/$/;
if ( !defined($mibDir) ) {
  if ( $opt->{mib_dir} eq '' ) {
    $mibDir = "${subAgentDir}mibs";
  } else { 
    $mibDir = $opt->{mib_dir};
  } 
} 
$config->{'log_to_console'} = 1 if $opt->{verbose};
undef $opt->{log_file} unless $opt->{log_file};

sub help {
  print <<EOHELP;
Version: @VERSION@

Usage: subagent-shell [options]

Options:
  -d.  --daemon             Work as daemon
  -p,  --pid-file=FILE      Set path to pid file (for daemon mode only)
  -b,  --base-dir=DIR       Path to installation root (default $opt->{base_dir})
  -m,  --mib-dir=DIR        Path to mib_dir (default $mibDir)
       --agentx-socket=FILE Path to agentx socket
  -l,  --log-file=FILE      Additionally log to a file 
  -v,  --verbose            Output varios debug information, 3 levels defined
  -e,  --exit-on-error      Test configuration at the initial run and stop if errors detected, 2 levels defined
  -h,  --help               Show this help information

EOHELP
}

if ($opt->{help}) {
  help;
  exit 0;
}

# ------ common functions ----------
sub unpackMessage {
  my $shm_message = shift;
  my $oData = shift;
  my $sOids = shift;
  my $prevTimeStamp = shift;

  my $curTimeStamp = unpack( 'S/A*', $shm_message );

  if ( ! $curTimeStamp ) {
      logMessage(LOG_NOTICE, "data is not processed yet.");  
      return 0;
  } elsif ( defined($$prevTimeStamp) && $curTimeStamp == $$prevTimeStamp ) {
      logMessage(LOG_DEBUG, "oid data is up to date, $curTimeStamp");  
      return 0;
  } 
  logMessage(LOG_DEBUG, "reading and unpacking data. prevTimeStamp: " . (defined($$prevTimeStamp)?$$prevTimeStamp:"") . ", curTimeStamp: $curTimeStamp");  
  my %shm_data;
  ( $$prevTimeStamp, %shm_data )= unpack( 'S/A* S/(S/A* S/A*)', $shm_message );

  my %typeMap = ( 'GAUGE'=>ASN_GAUGE,'TICKS'=>ASN_TIMETICKS,'INTEGER'=>ASN_INTEGER, 'INTEGER32'=>ASN_INTEGER, 'COUNTER'=>ASN_COUNTER, 'COUNTER64'=>ASN_COUNTER64, 'OCTETSTR'=>ASN_OCTET_STR, 'OPAQUE'=>ASN_OPAQUE, 'IPADDR'=>ASN_IPADDRESS, 'BITS'=>ASN_BIT_STR );
  %{$oData} = ();
  while ( my($k,$v) = each %shm_data) { 
    if ( $k !~ /^\./ ) { 
      # a mib variable with or without index
      my ( $mvar, $idx ) = split /\./, $k, 2; 
      if (my $node = $SNMP::MIB{$mvar}) { 
        $$oData{($node->{'objectID'}) . (defined($idx)?".$idx":'') }->{'V'} = $v;
        $$oData{($node->{'objectID'}) . (defined($idx)?".$idx":'') }->{'T'} = $typeMap{$node->{'type'}};
#        logMessage(LOG_DEBUG, "Type $node->{'objectID'} is $node->{'type'} map to $typeMap{$node->{'type'}}" ); 
      } else {
        logMessage(LOG_ERR, "$k is not found in a MIB, skipping"); 
      }
    } else {
      # an part of oid with without MIB
        $$oData{$reg_oid . '.107' . $k}->{'V'} = $v;
        $$oData{$reg_oid . '.107' . $k}->{'T'} = ASN_OCTET_STR;
    }
  }
  @$sOids = (sort { new NetSNMP::OID($a)<=>new NetSNMP::OID($b) } ( keys %$oData ) );
  return 0;
}

sub logMessage {
  my $level = shift;
  my $msg = shift;
  if ( $config->{'log_level'} && $config->{'log_level'} >= $level ) {
    syslog ( $level, $msg );
    print LOGFILE  strftime("%Y-%b-%d %H:%M:%S", localtime) . " $msg\n" if defined($opt->{log_file});
  }
  print STDERR "$msg\n" if defined($config->{'log_to_console'}) && $config->{'log_to_console'} != 0;
}

sub readConfig {
  logMessage(LOG_INFO, "Reading config ...");
  logMessage(LOG_DEBUG, "loading main config ${subAgentDir}subagent-shell-conf.xml");
  if ( ! -f "${subAgentDir}subagent-shell-conf.xml" ) {
    logMessage(LOG_ERR, "${subAgentDir}subagent-shell-conf.xml does not exists.Terminating");
    exit(1) 
  }
  my $x = XML::Simple->new();
  $config=$x->XMLin("${subAgentDir}subagent-shell-conf.xml");
  $config->{'log_to_console'} = 1 if $opt->{verbose};
  $agentMib{rootMIB}=$config->{'rootMIB'};
  $agentMib{functionsCount}=$config->{'functionsCount'};
  $agentMib{functionsSuccessCount}=$config->{'functionsSuccessCount'};
  $agentMib{functionsFailedCount}=$config->{'functionsFailedCount'};
  $agentMib{timeStamp}=$config->{'timeStamp'};
  $agentMib{totalExecTime}=$config->{'totalExecTime'};

# load extention functions
  foreach my $ff ( glob "${subAgentDir}*.functions" ) {
    logMessage(LOG_DEBUG, "loading functions from  $ff");
    require "$ff";
  }
# process hostname based configuration files
  my $h = hostname;
  foreach my $f_config ( (glob "${subAgentDir}subagent-shell-*-conf.xml"), (glob "${subAgentDir}conf.d/${h}*-conf.xml")) {
    logMessage(LOG_DEBUG, "loading functions config from $f_config");
    my $x = XML::Simple->new( ForceArray=>1, KeyAttr => '' );
    push @functions_config, $x->XMLin($f_config);
  }
# parse configuration and make the config to have 1 arg for a function. that's makes functions to run in paralel
  if ( defined($config->{'use_fork'}) && $config->{'use_fork'} eq 'yes' ) {
    my @add_config;
    foreach my $ca ( @functions_config ){
      foreach my $config_data ( @{$ca->{'function'}} ) {
        while ( defined($config_data->{'args'}) && scalar ( @{$config_data->{'args'}} ) > 1 )  {
          push @add_config ,  { 'function'=>[ { 'id'=>$config_data->{'id'}, 'args'=>[ pop @{$config_data->{'args'}} ] } ] };
        }
      }
    } 
    push @functions_config, @add_config;
  }
  if ( $opt->{verbose} > 0 ) {
    logMessage(LOG_DEBUG, "Configuration Dump:\n" . Data::Dumper->Dump( [ \@functions_config ] ));
  }
}

sub pollerInfo {
  my $o=shift;
  my $f=shift;
  my $s=shift;
  $$f{$agentMib{totalExecTime}} = sprintf('%.3f', scalar gettimeofday() - $s);
  while ( my($k,$v) = each  %$f ){ $$o{$k}=$v; }
  return 0;
} 

my %oData;
my @sortedoids;
my $dataTimeStamp;
sub pollerHandler {
  my ($handler, $registration_info, $request_info, $requests) = @_;

  sub oidToString {
    my $oid=shift;
    return "." . join (".", $oid->to_array());
  }
  sub unpackOData {
    my $r = shift;
    return ( defined($$r{'T'})?$$r{'T'}:ASN_OCTET_STR, $$r{'V'} );
  }

#read and unpack
  my $shm_message;
  semop($sem_id, pack("s!3", 0, 0, 0) . pack("s!3", 0, 1, 0) ); #lock
  unless (shmread($shm_id, $shm_message, 0, $config->{'shm_size'}-1)) { logMessage(LOG_ERR, "can't read shared memory at " . __LINE__); }
  semop($sem_id, pack("s!3", 0, -1, 0) ); #unlock
  eval {
    unpackMessage( $shm_message, \%oData, \@sortedoids, \$dataTimeStamp );
    1;
  } or do {
    logMessage(LOG_ERR, "exception caught in unpack message " . @$);
  };

  return if ($#sortedoids < 0 );

#  foreach my $k(@sortedoids) {
#    logMessage(LOG_DEBUG, "sorted oids $k -> $oData{$k}") ;
#  }
  for(my $request = $requests; $request; $request = $request->next()) {
    my $oid = $request->getOID();
    my $soid=oidToString($oid); 
    logMessage(LOG_DEBUG, "Requested: $oid" ); 
    if ( $request_info->getMode() == MODE_GET ) {
      logMessage(LOG_DEBUG, "MODE: GET $soid"); 
      if ( defined ( $oData{"$soid"} )) {
        $request->setValue(unpackOData($oData{$soid}));
      } 
    } elsif ( $request_info->getMode() == MODE_GETNEXT ) {
      logMessage(LOG_DEBUG, "MODE: GETNEXT, requested $oid"); 
      my $oididx=0;
      foreach (@sortedoids) { last if (new NetSNMP::OID($_) == $oid); $oididx++; }
      if ( $oid < new NetSNMP::OID($sortedoids[0]) ) {
        $request->setOID(new NetSNMP::OID($sortedoids[0]));
        $request->setValue(unpackOData($oData{$sortedoids[0]}));
      } elsif ( defined($sortedoids[$oididx+1]) && defined($oData{$sortedoids[$oididx+1]}) ) {
        $request->setOID(new NetSNMP::OID($sortedoids[$oididx+1]));
        $request->setValue(unpackOData($oData{$sortedoids[$oididx+1]}));
      } elsif (!defined( $oData{oidToString($oid)})) {
        logMessage(LOG_DEBUG, "LAST CHANCE requested $oid"); 
        my @nsoids=sort {new NetSNMP::OID($a)<=>new NetSNMP::OID($b)} (@sortedoids, oidToString($oid));
        $oididx=0;
        foreach (@nsoids) { last if (new NetSNMP::OID($_) == $oid); $oididx++; }
        if ( defined($nsoids[$oididx+1]) && defined($oData{$nsoids[$oididx+1]}) ) {
          $request->setOID(new NetSNMP::OID($nsoids[$oididx+1]));
          $request->setValue(unpackOData($oData{$nsoids[$oididx+1]}));
        }
      }
    } 
  }
}

sub cleanUp {
  shmctl($shm_id, IPC_RMID, 0);
  semctl($sem_id, 0, IPC_RMID, 0);
  msgctl($mQueue, IPC_RMID, 0) if $mQueue >= 0 ;
  unlink $ENV{SNMP_PERSISTENT_FILE} if -f $ENV{SNMP_PERSISTENT_FILE};
  unlink $opt->{pid_file} if defined($opt->{pid_file});
  close LOGFILE if defined($opt->{log_file});
  closelog();
}

sub daemonize {
  my $pid = fork();
  if (! defined $pid) { die "fork failed: $!"; }
  if ( $pid != 0 ) {
    closelog();
    close LOGFILE if defined($opt->{log_file});
    exit 0;
  }

  POSIX::setsid();

  if ($opt->{pid_file}) {
    if ( open PIDFILE, ">", $opt->{pid_file} ) {
      print PIDFILE "$$\n";
      close PIDFILE;
    } else {
      logMessage(LOG_WARNING, "Can't open pid-file $opt->{pid_file}, continuing without pid file.");
    }
  }
  eval {
    open STDIN,  "</dev/null" or die "Can't dup STDIN:  $!";
    open STDOUT, ">/dev/null" or die "Can't dup STDOUT: $!";
#    open(STDERR,"|/bin/logger -t \"subagent-shell\"[$$]") or die "Error: Unable to redirect STDERR to logger!";
    if ( defined($opt->{log_file})) {
      open(STDERR, ">&LOGFILE") or die "Error: Unable to redirect STDERR to log file!";
    } else {
      open STDERR, ">/dev/null" or die "Can't dup STDOUT: $!";
    }
    1;
  } or do {
    logMessage( LOG_ERR,"Can't dup $@" );
    exit 1;
  }
} 

# ===== main =====
{
  setlocale(LC_ALL,'C');      # avoid unexpected formatting
  $config->{'log_level'} = LOG_NOTICE; # set default logging until it's defined in config
  my $amode = ($0=~'snmp_perl.pl')?0:1;
  openlog( "subagent-shell", 'pid', LOG_DAEMON );
  if ( defined($opt->{log_file}) and not open( LOGFILE, ">>", $opt->{log_file} ) ){
    logMessage( LOG_WARNING, "Can't open log-file $opt->{log_file}, continuing without log file.");
    undef $opt->{log_file};
  }
  if ( $opt->{pid_file} && -f $opt->{pid_file} ) {
    open( PF, $opt->{pid_file} );
    my $p=<PF>; 
    close PF;
    chomp $p;
    if ( `ps hp $p -o cmd` =~ 'subagent-shell' ) {
      logMessage ( LOG_ERR, "subagent-shell process already running, exitting" ); 
      closelog();
      close LOGFILE if defined($opt->{log_file});
      exit(1)
    } else {
      logMessage (LOG_WARNING, "pid file already exists but process died" );
    }
  }  
  readConfig();

  my $agent = new NetSNMP::agent('Name' => "subagent-shell", 'AgentX' => $amode  );
  if ($opt->{agentx_socket} ne '' && $opt->{agentx_socket} ne '-') {
    netsnmp_ds_set_string(NETSNMP_DS_APPLICATION_ID, NETSNMP_DS_AGENT_X_SOCKET, $opt->{agentx_socket});
    logMessage (LOG_INFO, "AgentX socket set to $opt->{agentx_socket}");
  }
  SNMP::initMib(); # parses default list of Mib modules from default dirs
  SNMP::addMibDirs("$mibDir");

  if ( defined( $config->{'loadMibModules'} ) ) {
    if ( $config->{'loadMibModules'} =~ /all/i ) {
      logMessage(LOG_DEBUG, "loading all mib modules ");
      SNMP::loadModules("ALL");
    } elsif ( $config->{'loadMibModules'} =~ /defined/i ) {
      my @mibModules=("SNMPv2-SMI", "SNMPv2-TC", "$agentMib{rootMIB}-MIB");
      foreach ( glob("$mibDir/*MIB.txt") ){
        push @mibModules, (split /\./, basename($_) )[0];
      }
      logMessage(LOG_DEBUG, "loading mib modules: " . join ",", @mibModules);
      SNMP::loadModules(@mibModules);
    } else {
      my @mibModules=("SNMPv2-SMI", "SNMPv2-TC", "$agentMib{rootMIB}-MIB");
      push @mibModules, split /,/, $config->{'loadMibModules'};
      logMessage(LOG_DEBUG, "loading mib modules: " . join ",", @mibModules);
      SNMP::loadModules(@mibModules);
    }
  } else {
    logMessage(LOG_DEBUG, "loading all mib modules ");
    SNMP::loadModules("ALL");
  }

  $reg_oid=$SNMP::MIB{$agentMib{rootMIB}}->{'objectID'};

  if ( $opt->{daemon} ) {
    logMessage(LOG_WARNING, "can't become daemon in subagent mode") if ( $amode == 0 );
    daemonize();
  } 
  if ( $amode == 1 ) {
    $SIG{INT} = sub {
      $running = 0;
      logMessage(LOG_NOTICE, "Interrupted by user.");
    };
    $SIG{TERM} = sub {
      $running = 0;
      logMessage(LOG_NOTICE, "Terminated by user.");
    };
    $SIG{HUP} = sub {
      logMessage(LOG_NOTICE, "Received HUP, doing nothing.");
    };
  }

  unless (defined($mQueue=defined($config->{'use_fork'}) && $config->{'use_fork'} eq 'yes'?msgget(IPC_PRIVATE, IPC_CREAT | S_IRUSR | S_IWUSR):-1)) { logMessage(LOG_ERR, "can't msgget $!"); exit 1; };
  unless (defined($shm_id=shmget(IPC_PRIVATE, $config->{'shm_size'}, S_IRWXU)))   { logMessage(LOG_ERR, "can't shmget $!"); exit 1; }
  unless (defined($sem_id=semget($shm_id, 1, IPC_CREAT | 0700)))      { logMessage(LOG_ERR, "can't semctl $!"); exit 1; }
  unless (socketpair(CHILD, PARENT, AF_UNIX, SOCK_STREAM, PF_UNSPEC)) { logMessage(LOG_ERR, "can't make sockepair $!"); exit 1; };
  my $pid;
  my $parent_pid = $$; #to identify that the parent process is alive
  if ($pid = fork) {
# ===== parent ======
    close PARENT;
    unless ($agent)   { logMessage (LOG_ERR, 'Agent is not initialized'); exit 1; }
    unless ($reg_oid) { logMessage (LOG_ERR, 'Root OID is not configured'); exit 1; }
    unless ( $agent->register("subagent-shell", new NetSNMP::OID($reg_oid), \&pollerHandler ))
            { logMessage (LOG_ERR, 'failed to register agent'); exit 1; }
    logMessage (LOG_NOTICE, 'AgentX is initialized in ' . ($amode==0?'subagent':'standalone') . ' mode' ); 
    if ( $amode == 1) {
      $SIG{CHLD} = sub {
        while( waitpid(-1, WNOHANG) >0 ) {};
        logMessage(LOG_ERR, "Child process terminated or killed. Shutting down ...");
        $running = 0;
      };
      while($running) {
        eval {
          $agent->agent_check_and_process(1);
          1;
        } or do {
          logMessage(LOG_ERR,"exception caught in agent_check_and_process " . @$);
        };
      } 
      logMessage (LOG_NOTICE, 'AgentX shutdown' ); 
      $agent->shutdown();
      cleanUp();
    }
  } else { 
# ===== child =====
    $SIG{INT} = sub {
      logMessage(LOG_NOTICE, "SIGINT received, the parent process[$parent_pid] will be terminated.");
    };
    $SIG{TERM} = sub {
      logMessage(LOG_NOTICE, "SIGTERM received, the parent process[$parent_pid] will be terminated.");
    };
    close CHILD;
    unless ( defined( $pid ) )      { logMessage(LOG_ERR, "cannot fork: $!"); exit 1; }
    unless (open STDIN, ">&PARENT") { logMessage(LOG_ERR, "can't dup STDIN $!"); exit 1; }
    close PARENT;
    my %otree = (); 
    my %fStatus = (); 
    my %fStatusEx = (); 
    my $start_time;
    my $shm_message;
    my $cTested = 0;
    my $cTestFailed = 0;
    my %fHash = ();
    no strict "refs";
    
#extended information
    if ( defined($config->{'masterAgentPid'}) &&  defined($config->{'pollerPid'}) ) {
      $agentMib{masterAgentPid}=$config->{'masterAgentPid'};
      $agentMib{pollerPid}=$config->{'pollerPid'};
      %fStatusEx = ($agentMib{masterAgentPid}=>$parent_pid,
                    $agentMib{pollerPid}=$config->{'pollerPid'}=>$$);
    }

    logMessage(LOG_NOTICE, "Starting poller");

    while (1) { 
      logMessage(LOG_DEBUG, "polling with $config->{'poller_cycle'} s interval");
      %fStatus = ($agentMib{functionsCount}=>0, 
                  $agentMib{functionsSuccessCount}=>0, 
                  $agentMib{functionsFailedCount}=>0, 
                  $agentMib{timeStamp}=>strftime('%d-%b-%Y %H:%M:%S (%s)',localtime), 
                  $agentMib{totalExecTime}=>0);
      while ( my ($k,$v) = each(%fStatusEx) ) { $fStatus{$k} = $v; }

      %otree = (); 
      %fHash = ();
      my $curTimeStamp;
      my %func_pid = ();
      my $func_count = 1;
      my $s_time = scalar gettimeofday();

      foreach my $config_data ( @functions_config ){
        foreach my $func_config ( @{$config_data->{'function'}} ) {
          my $func = $func_config->{'id'};
          if ( defined( $fHash{$func} ) ) {
            $fHash{$func} ++;
          } else {
            $fHash{$func} = 1;
          }
          my $mCall =  $fHash{$func};
          if ( defined($config->{'use_fork'}) && $config->{'use_fork'} eq 'yes' ) {
            if ( $func_pid{$func . "($func_count)"} = fork ) {
            #    logMessage(LOG_INFO, "in parent for $func");
            # === parent ===
            $func_count ++;
            } else {
            # === child  ===
              %otree = ();
              eval {
                my $f_errno = &$func( \%otree, $func_config, \$mCall );
                my $msg =  pack( 'S/A* S S(S/A* S/A*)*', $func, $mCall , scalar keys( %otree ), %otree );
                utf8::downgrade($msg) || logMessage(LOG_ERR, "utf8::downgrade failed at " . __LINE__);
                msgsnd( $mQueue, pack("l! a*", length($msg), $msg), 0) || logMessage(LOG_ERR, "msgsnd failed for $func, error: $!. msglen:" . length($msg) . ". check /proc/sys/kernel/msgmax" ); 
                exit($f_errno);
              } or do {
                logMessage(LOG_WARNING, "function $func terminated abnormally");
                exit(242);
              };
            }
          } else {
            eval {
              if ( &$func( \%otree, $func_config, \$mCall ) == 0 ) { 
                $fStatus{$agentMib{functionsSuccessCount}} += 1;
              } else {
                $fStatus{$agentMib{functionsFailedCount}} += 1;
              } 
            } or do { 
              $fStatus{$agentMib{functionsFailedCount}} += 1;
              logMessage(LOG_WARNING, "function $func terminated abnormally");
            };
            $fStatus{$agentMib{functionsCount}} += 1;
            if ( $cTested == 0 && $fHash{$func} > 1 && $mCall != 0 ) {
                logMessage(LOG_ERR, "function $func defined $fHash{$func} times but does not support multicall.") ; 
                $cTestFailed = 1;
            }
          }
        }
      }

      if ( defined($config->{'use_fork'}) && $config->{'use_fork'} eq 'yes' ) {
        for my $f ( keys %func_pid ){
          local $SIG{ALRM} = sub { 
                              kill 9, $func_pid{$f}; 
                              logMessage(LOG_ERR, "function ${f}[$func_pid{$f}] timed out and killed ..."); 
                            };
          alarm $config->{'cmd_timeout'}+1;
          waitpid($func_pid{$f},0);
          alarm 0; 

          my $f_errno = $?;
          if ( $f_errno == 0 ) {
              $fStatus{$agentMib{functionsSuccessCount}} += 1;
          } else {
              $fStatus{$agentMib{functionsFailedCount}} += 1;
          }
          $fStatus{$agentMib{functionsCount}} += 1;
          if ( msgrcv( $mQueue, my $r_buf, $config->{'shm_size'}-1, 0, IPC_NOWAIT ) ) {
            my($r_len, $msg) = unpack( "l! a*", $r_buf );
            my ($func, $mCall, %r_otree ) =  unpack( 'S/A* S S/(S/A* S/A*)', $msg );
            %otree = ( %otree, %r_otree );

            if ( $cTested == 0 && $fHash{$func} > 1 && $mCall != 0 ) {
                logMessage(LOG_ERR, "function $func defined $fHash{$func} times but does not support multicall.") ; 
                $cTestFailed = 1;
            }

          }  else {
            logMessage(LOG_ERR, "a function did not send any message to queue. see previous erros.")  if ( $f_errno != 242 );
          }
        }
      }
  #   add pollerInfo
      pollerInfo(\%otree,\%fStatus,$s_time);
      $shm_message = pack( 'S/A* S(S/A* S/A*)*', (split /\(|\)/, $otree{$agentMib{timeStamp}})[1], scalar keys( %otree ), %otree );

      utf8::downgrade($shm_message, 1) || logMessage(LOG_ERR, "utf8::downgrade failed at " . __LINE__);

      if ( length($shm_message) >= $config->{'shm_size'} ) {
        logMessage(LOG_ERR, "data size " . length($shm_message) ." is greater than configured shm_size " . $config->{'shm_size'} );
        $cTestFailed = 1 if $cTested == 0;
      }

      if ( $opt->{verbose} >=1 ) {
        foreach my $k (sort keys %otree) { logMessage(LOG_DEBUG, "MIB from otree: $k => $otree{$k}" );}
      }
      if ( $opt->{verbose} >1 ) {
        my %oData;
        my $dataTimeStamp;
        my @sortedoids;
        unpackMessage( $shm_message, \%oData, \@sortedoids, \$dataTimeStamp );
        foreach my $o (@sortedoids) {
          my ( $t, $v ) = unpackOData($oData{$o});
          logMessage(LOG_DEBUG, "OIDS from unpackOData: $o = $t , $v" );
        }
      }

      if ( $cTested == 0 && ( $opt->{errors} >= 1 && $cTestFailed == 1 || $opt->{errors} >= 2 && $fStatus{$agentMib{functionsFailedCount}} > 0 ) ) {
        logMessage(LOG_ERR, "configuration test failed due to previous errors. Aborting ..." ); 
        exit(1) 
      }

      semop($sem_id, pack("s!3", 0, 0, 0) . pack("s!3", 0, 1, 0) ); #lock
      unless(shmwrite($shm_id, $shm_message, 0, $config->{'shm_size'}-1)) { logMessage(LOG_ERR, "can't write to shared memory at " . __LINE__); }
      semop($sem_id, pack("s!3", 0, -1, 0) ); #unlock

      my ($rin, $rout)  = ('', '');
      vec($rin, fileno(STDIN), 1) = 1;
      my ($nfound, $timeleft) = select($rout=$rin, undef, undef, $config->{'poller_cycle'});
      if ( $nfound != 0 ) {
        logMessage(LOG_NOTICE, "Connection with AgentX lost, shutdown poller.");
        cleanUp();
        exit 0; 
      }

      $cTested = 1;
    }
  }
}