dnxNebMain.c

Go to the documentation of this file.

/*--------------------------------------------------------------------------

   Copyright (c) 2006-2010, Intellectual Reserve, Inc. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License version 2 as
   published by the Free Software Foundation.

   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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

  --------------------------------------------------------------------------*/

#ifdef HAVE_CONFIG_H
# include "config.h"
#endif

#include "dnxNebMain.h"

#include "dnxCfgParser.h"
#include "dnxError.h"
#include "dnxDebug.h"

#ifndef NSCORE
# define NSCORE
#endif
#include "nagios.h"
#include "objects.h"    // for nagios service data type
#include "nebmodules.h"
#include "nebstructs.h"
#include "nebcallbacks.h"
#include "neberrors.h"
#include "broker.h"

#include <unistd.h>
#include <signal.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <regex.h>

#ifndef VERSION
# define VERSION    "<unknown>"
#endif

#ifndef SYSCONFDIR
# define SYSCONFDIR "/etc"
#endif

#ifndef SYSLOGDIR
# define SYSLOGDIR  "/var/log"
#endif

#ifndef LIBEXECDIR
# define LIBEXECDIR "/usr/libexec"
#endif

#if CURRENT_NEB_API_VERSION == 2
# define OBJECT_FIELD_NAME object
#elif CURRENT_NEB_API_VERSION == 3
# define OBJECT_FIELD_NAME object_ptr
#else
# error Unsupported NEB API version.
#endif

#define elemcount(x) (sizeof(x)/sizeof(*(x)))

#define DNX_DEFAULT_SERVER_CONFIG_FILE SYSCONFDIR "/dnxServer.cfg"
#define DNX_DEFAULT_LOG                SYSLOGDIR  "/dnxsrv.log"
#define DNX_DEFAULT_DBGLOG             SYSLOGDIR  "/dnxsrv.dbg.log"
#define DNX_DEFAULT_SERVER             LIBEXECDIR "/dnxServer"

NEB_API_VERSION(CURRENT_NEB_API_VERSION);

typedef struct DnxCfgData
{
   char * serverPath;         
   char * localCheckPattern;  
   char * syncScript;         
   char * logFilePath;        
   char * debugFilePath;      
   char * auditFilePath;      
   unsigned debugLevel;       
} DnxCfgData;

// module static data
static char * cfgfile;        
static DnxCfgData cfg;        
static DnxCfgParser * parser; 
static time_t start_time;     
static void * myHandle;       
static regex_t regEx;         
static pid_t serverpid;       
static int outdatafd;         
static int inackfd;           
static int indatafd;          
static pthread_t rlthread;    

/*--------------------------------------------------------------------------
                              IMPLEMENTATION
  --------------------------------------------------------------------------*/

static void releaseConfig(void)
{
   if (cfg.localCheckPattern)
      regfree(&regEx);

   dnxCfgParserDestroy(parser);
}

//----------------------------------------------------------------------------

static int validateCfg(DnxCfgDict * dict, void ** vptrs, void * passthru)
{
   regex_t * rep = (regex_t *)passthru;
   int err, ret = DNX_ERR_INVALID;
   DnxCfgData cfg;
   struct stat statbuf;

   assert(dict && vptrs && passthru);

   // setup data structure so we can use the same functionality we had before
   cfg.serverPath         = (char *)vptrs[ 0];
   cfg.localCheckPattern  = (char *)vptrs[ 1];
   cfg.syncScript         = (char *)vptrs[ 2];
   cfg.logFilePath        = (char *)vptrs[ 3];
   cfg.debugFilePath      = (char *)vptrs[ 4];
   cfg.debugLevel         = (unsigned)(intptr_t)vptrs[ 5];
   cfg.auditFilePath      = (char *)vptrs[ 6];

   // validate configuration items in context
   if (!cfg.serverPath || stat(cfg.serverPath, &statbuf) < 0
         || !S_ISREG(statbuf.st_mode))
      dnxLog("config: server not specified or invalid (\"%s\").",
             cfg.serverPath);
   else if (cfg.localCheckPattern && (err = regcomp(rep,
         cfg.localCheckPattern, REG_EXTENDED | REG_NOSUB)) != 0)
   {
      char buffer[128];
      regerror(err, rep, buffer, sizeof buffer);
      dnxLog("config: Failed to compile localCheckPattern (\"%s\"): %s.",
             cfg.localCheckPattern, buffer);
      regfree(rep);
   }
   else
      ret = 0;

   return ret;
}

//----------------------------------------------------------------------------

static int initConfig(char * cfgfile)
{
   DnxCfgDict dict[] =
   {  // Do NOT change the order, unless you know what you're doing!
      { "serverPath",         DNX_CFG_FSPATH,   &cfg.serverPath         },
      { "localCheckPattern",  DNX_CFG_STRING,   &cfg.localCheckPattern  },
      { "syncScript",         DNX_CFG_FSPATH,   &cfg.syncScript         },
      { "pluginLogFile",      DNX_CFG_FSPATH,   &cfg.logFilePath        },
      { "pluginDebugFile",    DNX_CFG_FSPATH,   &cfg.debugFilePath      },
      { "pluginDebugLevel",   DNX_CFG_UNSIGNED, &cfg.debugLevel         },
      { "pluginAuditFile",    DNX_CFG_FSPATH,   &cfg.auditFilePath      },
      { 0 },
   };
   char cfgdefs[] =
      "serverPath = " DNX_DEFAULT_SERVER "\n"
      "pluginLogFile = " DNX_DEFAULT_LOG "\n"
      "pluginDebugFile = " DNX_DEFAULT_DBGLOG "\n";

   int ret;
   regex_t re;

   // clear the regex string, as we may write into it
   memset(&re, 0, sizeof re);

   // create global configuration parser object
   if ((ret = dnxCfgParserCreate(cfgdefs, cfgfile, 0, dict,
         validateCfg, &parser)) != 0)
      return ret;

   // parse configuration file; pass defaults
   if ((ret = dnxCfgParserParse(parser, &re)) == 0)
      regEx = re;
   else
      dnxCfgParserDestroy(parser);

   return ret;
}

//----------------------------------------------------------------------------

static int nagiosGetServiceCount(void)
{
   extern service * service_list;      // the global nagios service list

   service * temp_service;
   int total_services = 0;

   // walk the service list, count the nodes
   for (temp_service = service_list; temp_service;
         temp_service = temp_service->next)
      total_services++;

   return total_services;
}

#if CURRENT_NEB_API_VERSION == 2

//----------------------------------------------------------------------------

static int nagios2xPostResult(service * svc, time_t start_time,
      int early_timeout, int res_code, char * res_data)
{
   extern circular_buffer service_result_buffer;
   int check_result_buffer_slots = 4096;

   service_message * new_message;

   // note that we're using malloc, not xmalloc - nagios takes ownership
   if ((new_message = (service_message *)malloc(sizeof *new_message)) == 0)
      return DNX_ERR_MEMORY;

   gettimeofday(&new_message->finish_time, 0);
   strncpy(new_message->host_name, svc->host_name, 
         sizeof(new_message->host_name) - 1);
   new_message->host_name[sizeof(new_message->host_name) - 1] = 0;
   strncpy(new_message->description, svc->description,
         sizeof(new_message->description) - 1);
   new_message->description[sizeof(new_message->description) - 1] = 0;
   new_message->return_code = res_code;
   new_message->exited_ok = TRUE;
   new_message->check_type = SERVICE_CHECK_ACTIVE;
   new_message->parallelized = svc->parallelize;
   new_message->start_time.tv_sec = start_time;
   new_message->start_time.tv_usec = 0L;
   new_message->early_timeout = early_timeout;
   strncpy(new_message->output, res_data, sizeof(new_message->output) - 1);
   new_message->output[sizeof(new_message->output) - 1] = 0;

   pthread_mutex_lock(&service_result_buffer.buffer_lock);

   // handle overflow conditions
   if (service_result_buffer.items == check_result_buffer_slots)
   {
      service_result_buffer.overflow++;
      service_result_buffer.tail = (service_result_buffer.tail + 1)
            % check_result_buffer_slots;
   }

   // save the data to the buffer
   ((service_message **)service_result_buffer.buffer)
         [service_result_buffer.head] = new_message;

   // increment the head counter and items
   service_result_buffer.head = (service_result_buffer.head + 1)
         % check_result_buffer_slots;
   if (service_result_buffer.items < check_result_buffer_slots)
      service_result_buffer.items++;
   if (service_result_buffer.items > service_result_buffer.high)
      service_result_buffer.high = service_result_buffer.items;

   pthread_mutex_unlock(&service_result_buffer.buffer_lock);

   return 0;
}

#elif CURRENT_NEB_API_VERSION == 3

// comment out this line to post to N3 filesystem results queue
#define DIRECT_POST

#ifdef DIRECT_POST

// a nagios core global (It shouldn't be global, but we're glad it is!)
extern check_result * check_result_list;

static check_result * dnxResultList = 0;
static pthread_mutex_t dnxResultListMutex = PTHREAD_MUTEX_INITIALIZER;

//----------------------------------------------------------------------------

static long dnxTimeCompare(struct timeval * tv1, struct timeval * tv2)
{
   long secdiff = (long)(tv1->tv_sec - tv2->tv_sec);
   long usecdiff = (long)(tv1->tv_usec - tv2->tv_usec);
   return secdiff? secdiff: usecdiff;
}

//----------------------------------------------------------------------------

static void dnxAddResultToList(check_result * newcr)
{
   check_result ** curp;

   assert(newcr);

   DNX_PT_MUTEX_LOCK(&dnxResultListMutex);

   for (curp = &dnxResultList; *curp; curp = &(*curp)->next)
      if (dnxTimeCompare(&(*curp)->finish_time, &newcr->finish_time) >= 0)
         break;

   newcr->next = *curp;
   *curp = newcr;

   DNX_PT_MUTEX_UNLOCK(&dnxResultListMutex);
}

//----------------------------------------------------------------------------

static int nagios3xPostResult(service * svc, int check_type,
      int check_options, int schedule, int reschedule, 
      time_t start_time, time_t finish_time, int early_timeout,
      int exited_ok, int res_code, char * res_data)
{
   check_result * newcr;

   // use malloc here, not xmalloc - nagios will free it
   if ((newcr = (check_result *)malloc(sizeof *newcr)) == 0)
      return DNX_ERR_MEMORY;

   // initialize and fill with result info
   newcr->object_check_type = SERVICE_CHECK;
   newcr->host_name = strdup(svc->host_name);
   newcr->service_description = strdup(svc->description);
   newcr->check_type = check_type;
   newcr->check_options = check_options;
   newcr->scheduled_check = schedule;
   newcr->reschedule_check = reschedule;
   newcr->output_file = 0;
   newcr->output_file_fd = -1;
   newcr->latency = svc->latency;
   newcr->start_time.tv_sec = start_time;
   newcr->start_time.tv_usec = 0;
   newcr->finish_time.tv_sec = finish_time;
   newcr->finish_time.tv_usec = 0;
   newcr->early_timeout = early_timeout;
   newcr->exited_ok = exited_ok;
   newcr->return_code = res_code;
   newcr->output = strdup(res_data);
   newcr->next = 0;

   dnxAddResultToList(newcr);

   return 0;
}

//----------------------------------------------------------------------------

static check_result * dnxMergeLists(check_result * lista, check_result * listb)
{
   check_result * result = 0;

   check_result ** iter;
   for (iter = &result; lista && listb; iter = &(*iter)->next)
      if (dnxTimeCompare(&lista->finish_time, &listb->finish_time) <= 0)
      { *iter = lista; lista = lista->next; }
      else
      {*iter = listb; listb = listb->next; }

   *iter = lista? lista: listb;

   return result;
}

//----------------------------------------------------------------------------

static void dnxMoveResultsToNagios(void)
{
   check_result * local;

   // safely save off currently local list
   DNX_PT_MUTEX_LOCK(&dnxResultListMutex);
   local = dnxResultList;
   dnxResultList = 0;
   DNX_PT_MUTEX_UNLOCK(&dnxResultListMutex);

   // merge local into check_result_list, store in check_result_list
   check_result_list = dnxMergeLists(local, check_result_list);
}

//----------------------------------------------------------------------------

static int ehTimedEvent(int event_type, void * data)
{
   nebstruct_timed_event_data * ted = (nebstruct_timed_event_data *)data;
   timed_event * event = (timed_event*)data;
   int ret;

   // sanity checks
   if (event_type != NEBCALLBACK_TIMED_EVENT_DATA || ted == 0)
      return ERROR;

   // we only care about REAPER events
   if (ted->event_type != EVENT_CHECK_REAPER)
      return OK;

   dnxDebug(3, "Reaper handler called.");
   
   dnxMoveResultsToNagios();

   return OK;
}

#else // !DIRECT_POST

//----------------------------------------------------------------------------

static void nagios3xMoveCheckResultToQueue(char * resfile)
{
   char * outfile;
   int ret = -1;

   // a nagios 3x global variable
   extern char * check_result_path;

   // create a safe temp file
   // path length + 'cXXXXXX' (7) + '/' (1) + '.ok' (3) + null (1)
   if ((outfile = (char *)xmalloc(strlen(check_result_path) 
         + 7 + 1 + 3 + 1)) != 0)
   {
      int fd;
      //int oldmask = umask(077);

      sprintf(outfile, "%s/cXXXXXX", check_result_path);
      if ((fd = mkstemp(outfile)) > 0)
      {
#ifdef __CYGWIN__
         // Cygwin cannot rename open files - gives Permission Denied
         close(fd);
#endif

         // move the original file using a nagios core function
         ret = my_rename(resfile, outfile);

#ifndef __CYGWIN__
         // close the file
         close(fd);
#endif

         // create an ok-to-go indicator file
         strcat(outfile, ".ok");
         if ((fd = open(outfile, O_CREAT|O_WRONLY|O_TRUNC, S_IRUSR|S_IWUSR)) > 0)
            close(fd);

      }

      //umask(oldmask);

      xfree(outfile);
   }

   // delete the original file if allocation failed or it couldn't be moved
   if (ret != 0)
      unlink(resfile);
}

//----------------------------------------------------------------------------

static int nagios3xPostResult(service * svc, int check_type,
      int check_options, int schedule, int reschedule, 
      time_t start_time, time_t finish_time, int early_timeout,
      int exited_ok, int res_code, char * res_data)
{
   char * filename;
   FILE * fp = 0;
   int fd;

   // open a file in the check results path for storing check results
   // path length + 'checkXXXXXX' (11) + '/' (1) + '.ok' (3) + null (1)
   if ((filename = (char *)xmalloc(strlen(temp_path) 
         + 11 + 1 + 3 + 1)) == 0)
      return DNX_ERR_MEMORY;

   sprintf(filename, "%s/checkXXXXXX", temp_path);
   if ((fd = mkstemp(filename)) < 0
         || (fp = fdopen(fd, "w")) == 0)
   {
      xfree(filename);
      if (fd >= 0) close(fd);
      return DNX_ERR_OPEN;
   }

   // write check results to the queue file
   fprintf(fp, "### Active Check Result File ###\n");
   fprintf(fp, "file_time=%lu\n\n", (unsigned long)start_time);
   fprintf(fp, "### Nagios Service Check Result ###\n");
   fprintf(fp, "# Time: %s", ctime(&start_time));
   fprintf(fp, "host_name=%s\n", svc->host_name);
   fprintf(fp, "service_description=%s\n", svc->description);
   fprintf(fp, "check_type=%d\n", check_type);
   fprintf(fp, "check_options=%d\n", check_options);
   fprintf(fp, "scheduled_check=%d\n", schedule);
   fprintf(fp, "reschedule_check=%d\n", reschedule);
   fprintf(fp, "latency=%f\n", svc->latency);
   fprintf(fp, "start_time=%lu.0\n", (unsigned long)start_time);
   fprintf(fp, "finish_time=%lu.%lu\n", (unsigned long)finish_time);
   fprintf(fp, "early_timeout=%d\n", early_timeout);
   fprintf(fp, "exited_ok=%d\n", exited_ok);
   fprintf(fp, "return_code=%d\n", res_code);
   fprintf(fp, "output=%s\n", res_data);

   fclose(fp);

   nagios3xMoveCheckResultToQueue(filename);

   xfree(filename);

   return 0;
}

#endif // ?DIRECT_POST

#else
# error Unsupported NEB API version.
#endif

//----------------------------------------------------------------------------

static int dnxPostResult(void * data, time_t start_time, unsigned delta,
      int early_timeout, int res_code, char * res_data)
{
   DnxJobTransfer * job = (DnxJobTransfer *)data;
   service * svc = (service *)job->svc;
   int ret;

   assert(job);

   if (early_timeout)
      res_code = STATE_UNKNOWN;

#if CURRENT_NEB_API_VERSION == 2

   ret = nagios2xPostResult(svc, start_time, early_timeout,
         res_code, res_data);

#elif CURRENT_NEB_API_VERSION == 3

   ret = nagios3xPostResult(svc, SERVICE_CHECK_ACTIVE,
         job->chkopts, job->schedule, job->reschedule,
         start_time, start_time + delta, early_timeout,
         1, res_code, res_data);

#else
# error Unsupported NEB API version.
#endif

   xfree(job); // free the originally allocated DnxJobTransfer structure

   return ret;
}

//----------------------------------------------------------------------------

static int GetMsgHeader(DnxMsgHeader * phdr)
{
   size_t toread = sizeof *phdr;
   char * bufptr = (char *)phdr;
   ssize_t bytesread;

   // read the message header from the plugin module
   while (toread > 0 && (bytesread = read(indatafd, bufptr, toread)) > 0)
   {
      toread -= bytesread;
      bufptr += bytesread;
   }

   // check for closed pipe
   if (bytesread == 0)
      return EPIPE;

   // check for error
   if (bytesread < 0)
   {
      int ret = errno;

      dnxDebug(1, "GetMsgHeader: read failed: %s.", strerror(ret));
      return ret;
   }

   // sanity check message format
   if (phdr->signature != DNX_MSG_SIGNATURE
         || phdr->structsz > DNX_MAX_XFER_SIZE)
   {
      dnxDebug(1, "GetMsgHeader: unsupported message format.");
      return DNX_ERR_UNSUPPORTED;
   }

   return 0;
}

//----------------------------------------------------------------------------

static int ProcessResultTransferMsg(DnxResultTransfer * result)
{
   size_t toread = result->hdr.structsz - sizeof result->hdr;
   char * bufptr = (char *)result + sizeof result->hdr;
   ssize_t bytesread;
   int ret;

   // read the rest of the result transfer structure
   while (toread > 0 && (bytesread = read(indatafd, bufptr, toread)) > 0)
   {
      bufptr += bytesread;
      toread -= bytesread;
   }

   // check for closed pipe
   if (bytesread == 0)
      return EPIPE;

   // check for error 
   if (bytesread < 0)
   {
      ret = errno;

      dnxDebug(1, "ProcessResultTransferMsg: read failed: %s.", strerror(ret));
      return ret;
   }

   // post result to Nagios
   if ((ret = dnxPostResult(result->payload, result->start_time, result->delta,
         result->early_timeout, result->res_code, result->res_data)) != 0)
      dnxLog("Failed to post result [%lu] to Nagios; \"%s\": %d.", 
            result->serial, result->res_data, ret);

   return ret;
}

//----------------------------------------------------------------------------

static void * dnxResultsListener(void * arg) 
{
   DnxResultTransfer * result = 0;
   size_t resultsz = 0;
   int terminating = 0;

   // ignore broken pipe signals to this thread
   signal(SIGPIPE, SIG_IGN);
   
   dnxLog("Results listener thread awaiting results...");

   while (!terminating)
   {
      int ret;
      DnxMsgHeader hdr;
      if ((ret = GetMsgHeader(&hdr)) == 0)
      {
         switch (hdr.msgtype)
         {
            case DNX_MSG_RESULT_TRANSFER:
               // resize result transfer buffer if necessary
               if (hdr.structsz > resultsz)
               {
                  DnxResultTransfer * tmp;
                  if ((tmp = (DnxResultTransfer *)xrealloc(result, hdr.structsz)) != 0)
                  {
                     result = tmp;
                     resultsz = hdr.structsz;
                  }
               }

               // copy header and read rest of job transfer message
               result->hdr = hdr;
               ret = ProcessResultTransferMsg(result);
               break;
         }
      }
      if (ret == EPIPE)
         terminating = 1;
   }

   dnxLog("Results listener thread terminating...");

   xfree(result);
   return 0;
}

//----------------------------------------------------------------------------

static int GetAckNak(void)
{
   int ret;
   size_t toread = sizeof ret;
   char * bufptr = (char *)&ret;
   ssize_t bytesread;

   while (toread > 0 && (bytesread = read(inackfd, bufptr, toread)) > 0)
   {
      bufptr += bytesread;
      toread -= bytesread;
   }

   // check for closed pipe
   if (bytesread == 0)
      return EPIPE;

   // check for error
   if (bytesread < 0)
   {
      ret = errno;

      dnxDebug(1, "GetAckNak: read failed: %s.", strerror(ret));
      return ret;
   }

   return ret;
}

//----------------------------------------------------------------------------

static int AllocateNodeRequest(void)
{
   DnxMsgHeader nodereq;
   size_t towrite = nodereq.structsz = sizeof nodereq;
   char * bufptr = (char *)&nodereq;
   ssize_t byteswritten;
   int ret;

   dnxDebug(3, "Allocating node request.");

   // allocate client node request - bail on EAGAIN - pipe full, no one's listening
   nodereq.signature = DNX_MSG_SIGNATURE;
   nodereq.msgtype = DNX_MSG_RESERVE_NODEREQ;

   while (towrite > 0 && (byteswritten = write(outdatafd, bufptr, towrite)) > 0)
   {
      bufptr += byteswritten;
      towrite -= byteswritten;
   }

   // check for closed pipe
   if (byteswritten == 0)
      return EPIPE;

   // check for error
   if (byteswritten < 0)
   {
      ret = errno;

      dnxDebug(1, "Alloc node request: write failed: %s.", strerror(ret));
      return ret;
   }

   return GetAckNak();
}

//----------------------------------------------------------------------------

static int TransferJobToServer(DnxJobTransfer * job)
{
   char * bufptr = (char *)job;
   size_t towrite = job->hdr.structsz;
   ssize_t byteswritten;
   int ret;

   assert(job);

   dnxDebug(3, "Transferring service check.");

   // write job to server - bail on EAGAIN - pipe full, no one's listening
   job->hdr.signature = DNX_MSG_SIGNATURE;
   job->hdr.msgtype = DNX_MSG_JOB_TRANSFER;
   while (towrite > 0 && (byteswritten = write(outdatafd, bufptr, towrite)) > 0)
   {
      bufptr += byteswritten;
      towrite -= byteswritten;
   }

   // check for closed socket
   if (byteswritten == 0)
      return EPIPE;

   // check for error
   if (byteswritten < 0)
   {
      ret = errno;

      dnxDebug(1, "Alloc node request: write failed: %s.", strerror(ret));
      return ret;
   }

   return GetAckNak();
}

//----------------------------------------------------------------------------

static int dnxPostNewJob(unsigned long serial, nebstruct_service_check_data * ds)
{
   service * svc = (service *)ds->OBJECT_FIELD_NAME;
   DnxJobTransfer * job;
   size_t jobsz;
   int ret;

   assert(ds);
   assert(svc);
   assert(ds->command_line);

   // tell server to allocate a node request
   if ((ret = AllocateNodeRequest()) != 0)
      return ret;

   // allocate a job transfer structure
   jobsz = sizeof(*job) + strlen(ds->command_line);
   if ((job = (DnxJobTransfer *)xmalloc(jobsz)) == 0)
      return DNX_ERR_MEMORY;

   job->hdr.structsz = jobsz;
   job->serial = serial;
   job->payload = job;  // track job transfer structure between job and result
   job->svc = svc;      // track service for use in result posting 
   job->start_time = ds->start_time.tv_sec;
   job->timeout = ds->timeout;
   job->expires = job->start_time + job->timeout + 5;

#if CURRENT_NEB_API_VERSION == 3
   {
      // a nagios 3.x global variable
      extern check_result check_result_info;

      job->chkopts = check_result_info.check_options;
      job->schedule = check_result_info.scheduled_check;
      job->reschedule = check_result_info.reschedule_check;
   }
#endif

   strcpy(job->cmd, ds->command_line);

   return TransferJobToServer(job);
} 

//----------------------------------------------------------------------------

static int ehSvcCheck(int event_type, void * data)
{
   static unsigned long serial = 0;

   nebstruct_service_check_data * svcdata = (nebstruct_service_check_data *)data;
   int ret;

   // shouldn't happen because we didn't sign up for others
   if (event_type != NEBCALLBACK_SERVICE_CHECK_DATA)
      return OK;

   // if this happens it's a bug in Nagios
   if (svcdata == 0)
   {
      dnxDebug(1, "Service handler received NULL service data structure.");
      return ERROR;  // shouldn't happen - internal Nagios error
   }

   // we only want the first BROKER event we get on each check
   if (svcdata->type != NEBTYPE_SERVICECHECK_INITIATE)
      return OK;     // ignore non-initiate service checks

   // every valid service check is assigned a unique serial number
   serial++;

   dnxDebug(3, "Service handler received service check [%lu] at %lu.",
         serial, (unsigned long)svcdata->start_time.tv_sec);

   // check for local execution pattern on command line
   if (cfg.localCheckPattern 
         && regexec(&regEx, svcdata->command_line, 0, 0, 0) == 0)
   {
      dnxDebug(1, "Regex filter. Service check [%lu] will execute locally: %s.", 
            serial, svcdata->command_line);
      return OK;
   }

   // try to post this job to the dnx server process
   if ((ret = dnxPostNewJob(serial, svcdata)) != DNX_OK)
   {
      dnxDebug(1, "Post failed: %s. Service check [%lu] will execute locally: %s.", 
            dnxErrorString(ret), serial, svcdata->command_line);
      return OK;
   }

   return NEBERROR_CALLBACKOVERRIDE;   // tell nagios we want it
}

//----------------------------------------------------------------------------

static int dnxCalculateJobListSize(void)
{
   int size = nagiosGetServiceCount();

   // zero doesn't make sense...
   if (size < 1)
   {
      size = 100;
      dnxLog("No Nagios services defined! "
             "Defaulting to %d slots in the DNX job queue.", size);
   }
   return size;
}

//----------------------------------------------------------------------------

static int execServerProcess(void)
{

#define P0R 0     // Pipe 0 Read descriptor
#define P0W 1     // Pipe 0 Write descriptor
#define P1R 2     // Pipe 1 Read descriptor
#define P1W 3     // Pipe 1 Write descriptor
#define P2R 4     // Pipe 2 Read descriptor
#define P2W 5     // Pipe 2 Write descriptor
#define PMX 6     // Pipe Max desscriptors

   pid_t pid;
   int flags, fds[PMX];

   // open three IPC pipes and fork the process
   memset(fds, -1, sizeof fds);
   if (pipe(fds + P0R) < 0 || pipe(fds + P1R) < 0 
         || pipe(fds + P2R) < 0 || (pid = fork()) < 0)
   {
      int i, ret = errno;

      for (i = 0; i < elemcount(fds); i++)
         if (fds[i] >= 0) 
            close(fds[i]);

      return ret;
   }

   // *** zero is child process ***

   if (pid == 0)
   {
      int svclistsz = dnxCalculateJobListSize();
      char strsvclistsz[16];

      // create a string version of the service list size
      snprintf(strsvclistsz, sizeof strsvclistsz, "%d", svclistsz);

      close(fds[P0W]);     // job xfer    - server closes write end
      close(fds[P1R]);     // ack/nak     - server closes read end
      close(fds[P2R]);     // result xfer - server closes read end
      
      // dup open pipe descriptors onto well-known descriptors, if needed
      if (fds[P0R] != DNX_SRVFD)
      {
         dup2(fds[P0R], DNX_SRVFD);
         close(fds[P0R]); 
      }
      if (fds[P1W] != DNX_SAKFD)
      {
         dup2(fds[P1W], DNX_SAKFD);
         close(fds[P1W]);
      }
      if (fds[P2W] != DNX_PLGFD)
      {
         dup2(fds[P2W], DNX_PLGFD);
         close(fds[P2W]);
      }

      // exec DNX Server process, pass config args
      execl(cfg.serverPath, cfg.serverPath, "-c", cfgfile, 
            "-j", strsvclistsz, (char *)0);

      _exit(1);            // error in exec - kill child

      // *** NO RETURN FROM HERE ***
   }

   // *** otherwise we're still in the parent process ***

   close(fds[P0R]);        // job xfer    - plugin closes read end
   close(fds[P1W]);        // ack/nak     - plugin closes write end
   close(fds[P2W]);        // result xfer - plugin closes write end

   // save open pipe descriptors in global variables
   outdatafd = fds[P0W];
   inackfd   = fds[P1R];
   indatafd  = fds[P2R];

   // set non-block flag on plugin -> server write channel
   // if the pipe is full, we'll let Nagios handle the check
   if ((flags = fcntl(outdatafd, F_GETFL, 0)) >= 0)
   {
      flags |= O_NONBLOCK;
      fcntl(outdatafd, F_SETFL, flags);
   }

   // save server pid for wait call
   serverpid = pid;

   return 0;
}

//----------------------------------------------------------------------------

static int terminateServerProcess(void)
{
   int ret = 0;
   if (serverpid > 0)
   {
      kill(serverpid, SIGTERM);  // sigterm stops agent thread
      close(outdatafd);          // closing pipes stop request handler
      close(inackfd);
      if (waitpid(serverpid, 0, 0) < 0)
         ret = errno;
   }
   return ret;
}

//----------------------------------------------------------------------------

// forward declaration due to circular reference
static int ehProcessData(int event_type, void * data);

static int dnxPluginDeInit(void)
{
   // deregister for all nagios events we previously registered for...
   neb_deregister_callback(NEBCALLBACK_PROCESS_DATA, ehProcessData);
   neb_deregister_callback(NEBCALLBACK_SERVICE_CHECK_DATA, ehSvcCheck);
#if CURRENT_NEB_API_VERSION == 3 && defined(DIRECT_POST)
   neb_deregister_callback(NEBCALLBACK_TIMED_EVENT_DATA, ehTimedEvent);
#endif

   // shutdown server child process
   if (serverpid != 0)
      terminateServerProcess();

   // terminate results listener thread
   if (rlthread != 0)
   {
      close(indatafd);           // closing pipe signals thread
      pthread_join(rlthread, 0);
   } 

   dnxLog("DNX Nagios Plugin shutdown completed.");

   return OK;  // Nagios OK value
}

//----------------------------------------------------------------------------

static int dnxPluginInit(void)
{
   int ret;

   // start dnx server child process
   if ((ret = execServerProcess()) != 0)
   {
      dnxLog("Error starting DNX server process: %s.", dnxErrorString(ret));
      return ret;
   }

   dnxLog("DNX Server process started.");

   // start results listener
   if ((ret = pthread_create(&rlthread, 0, dnxResultsListener, (void *)0)) != 0)
      return ret;
   
   dnxLog("Results listener thread started.");

#if CURRENT_NEB_API_VERSION == 3 && defined(DIRECT_POST)

   // register for timed event to piggy-back on reaper thread
   neb_register_callback(NEBCALLBACK_TIMED_EVENT_DATA, myHandle, 0, ehTimedEvent);
   dnxLog("Registered for TIMEDEVENT_EXECUTE event.");

#endif

   // registration for this event starts everything rolling
   neb_register_callback(NEBCALLBACK_SERVICE_CHECK_DATA, myHandle, 0, ehSvcCheck);
   dnxLog("Registered for SERVICE_CHECK_DATA event.");

   dnxLog("DNX Nagios Plugin initialization completed.");

   return OK;  // Nagios OK value
}

//----------------------------------------------------------------------------

static int launchScript(char * script)
{
   int ret;

   assert(script);

   // exec the script - system waits till child completes
   if ((ret = system(script)) == -1)
   {
      dnxLog("Failed to exec script: %s.", strerror(errno));
      ret = DNX_ERR_INVALID;
   }
   else
      ret = DNX_OK;

   dnxLog("Sync script returned %d.", WEXITSTATUS(ret));

   return ret;
}

//----------------------------------------------------------------------------

static int ehProcessData(int event_type, void * data)
{
   nebstruct_process_data *procdata = (nebstruct_process_data *)data;

   // validate our event type - ignore wrong event type
   assert(event_type == NEBCALLBACK_PROCESS_DATA);
   if (event_type != NEBCALLBACK_PROCESS_DATA)
      return OK;

   // sanity-check our data structure - should never happen
   if (!procdata)
   {
      dnxLog("Startup handler received NULL process data structure.");
      return ERROR;
   }

   // look for process event loop start event
   if (procdata->type == NEBTYPE_PROCESS_EVENTLOOPSTART)
   {
      dnxDebug(2, "Startup handler received PROCESS_EVENTLOOPSTART event.");

      // execute sync script, if defined
      if (cfg.syncScript)
      {
         dnxLog("Startup handler executing plugin sync script: %s.", cfg.syncScript);

         // NB: This halts Nagios execution until the script exits...
         launchScript(cfg.syncScript);
      }

      // if plugin init fails, do plugin shutdown
      if (dnxPluginInit() != 0)
         dnxPluginDeInit();
   }
   return OK;
}

/*--------------------------------------------------------------------------
                                 INTERFACE
  --------------------------------------------------------------------------*/

//----------------------------------------------------------------------------

int nebmodule_deinit(int flags, int reason)
{
   dnxLog("-------- DNX Nagios Plugin Module Shutdown Initiated --------");
   
   dnxPluginDeInit();
   releaseConfig();
   xheapchk();

   dnxLog("-------- DNX Nagios Plugin Module Shutdown Completed --------");
   return 0;
}

//----------------------------------------------------------------------------

int nebmodule_init(int flags, char * args, nebmodule * handle)
{
   int ret;

   myHandle = handle;

   // module args string should contain a fully-qualified config file path
   if (!args || !*args)
      args = DNX_DEFAULT_SERVER_CONFIG_FILE;

   cfgfile = args;
   if ((ret = initConfig(cfgfile)) != 0)
      return ERROR;

   // set configured debug level and syslog log facility code
   dnxLogInit(cfg.logFilePath, cfg.debugFilePath, cfg.auditFilePath, 
         &cfg.debugLevel);

   dnxLog("-------- DNX Nagios Plugin Module Version %s Startup --------", VERSION);
   dnxLog("Copyright (c) 2006-2010 Intellectual Reserve. All rights reserved.");
   if (cfg.debugLevel)
      dnxLog("Debug logging enabled at level %d to %s.",
            cfg.debugLevel, cfg.debugFilePath);
   if (cfg.auditFilePath)
      dnxLog("Auditing enabled to %s.", cfg.auditFilePath);

   // subscribe to PROCESS_DATA call-backs in order to defer initialization
   //    until after Nagios validates its configuration and environment.
   if ((ret = neb_register_callback(NEBCALLBACK_PROCESS_DATA,
         myHandle, 0, ehProcessData)) != OK)
   {
      dnxLog("PROCESS_DATA event registration failed: %s.", dnxErrorString(ret));
      terminateServerProcess();
      releaseConfig();
      return ERROR;
   }
   start_time = time(0);

   dnxLog("-------- DNX Nagios Plugin Module Startup Complete --------");

   return OK;
}