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

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

#include "dnxNebMain.h"
#include "dnxError.h"
#include "dnxDebug.h"
#include "dnxLogging.h"
#include "dnxCollector.h"
#include "dnxDispatcher.h"
#include "dnxRegistrar.h"
#include "dnxJobList.h"
#include "dnxCfgParser.h"
#include "dnxStats.h"
#include "dnxAgent.h"

#ifdef HAVE_CONFIG_H
# include "config.h"
#else
# define VERSION "<unknown>"
#endif

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

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

#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 <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <regex.h>

#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"

// specify event broker API version (required)
NEB_API_VERSION(CURRENT_NEB_API_VERSION);

typedef struct DnxJobData
{
   service * svc;                   
   int chkopts;                     
   int schedule;                    
   int reschedule;                  
} DnxJobData;

typedef struct DnxServerCfg
{
   char * agentUrl;                 
   char * dispatcherUrl;            
   char * collectorUrl;             
   char * authWorkerNodes;          
   unsigned maxNodeRequests;        
   unsigned minServiceSlots;        
   unsigned expirePollInterval;     
   char * localCheckPattern;        
   char * syncScript;               
   char * logFilePath;              
   char * debugFilePath;            
   char * auditFilePath;            
   unsigned debugLevel;             
} DnxServerCfg;

// module static data
static DnxServerCfg cfg;            
static DnxCfgParser * parser;       
static DnxJobList * joblist;        
static DnxRegistrar * registrar;    
static DnxDispatcher * dispatcher;  
static DnxCollector * collector;    
static time_t start_time;           
static int s_shutdown;              
static DnxChannel * agent;          
static pthread_t agentTid;          
static void * myHandle;             
static regex_t regEx;               

/*--------------------------------------------------------------------------
                              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;
   DnxServerCfg cfg;

   assert(dict && vptrs && passthru);

   // setup data structure so we can use the same functionality we had before
   cfg.agentUrl           = (char *)vptrs[ 0];
   cfg.dispatcherUrl      = (char *)vptrs[ 1];
   cfg.collectorUrl       = (char *)vptrs[ 2];
   cfg.authWorkerNodes    = (char *)vptrs[ 3];
   cfg.maxNodeRequests    = (unsigned)(intptr_t)vptrs[ 4];
   cfg.minServiceSlots    = (unsigned)(intptr_t)vptrs[ 5];
   cfg.expirePollInterval = (unsigned)(intptr_t)vptrs[ 6];
   cfg.localCheckPattern  = (char *)vptrs[ 7];
   cfg.syncScript         = (char *)vptrs[ 8];
   cfg.logFilePath        = (char *)vptrs[ 9];
   cfg.debugFilePath      = (char *)vptrs[10];
   cfg.auditFilePath      = (char *)vptrs[11];
   cfg.debugLevel         = (unsigned)(intptr_t)vptrs[12];

   // validate configuration items in context
   if (!cfg.agentUrl)
      dnxLog("config: Missing channelAgent parameter.");
   else if (!cfg.dispatcherUrl)
      dnxLog("config: Missing channelDispatcher parameter.");
   else if (!cfg.collectorUrl)
      dnxLog("config: Missing channelCollector parameter.");
   else if (cfg.maxNodeRequests < 1)
      dnxLog("config: Invalid maxNodeRequests parameter.");
   else if (cfg.minServiceSlots < 1)
      dnxLog("config: Invalid minServiceSlots parameter.");
   else if (cfg.expirePollInterval < 1)
      dnxLog("config: Invalid expirePollInterval parameter.");
   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!
      { "channelAgent",       DNX_CFG_URL,      &cfg.agentUrl           },
      { "channelDispatcher",  DNX_CFG_URL,      &cfg.dispatcherUrl      },
      { "channelCollector",   DNX_CFG_URL,      &cfg.collectorUrl       },
      { "authWorkerNodes",    DNX_CFG_STRING,   &cfg.authWorkerNodes    },
      { "maxNodeRequests",    DNX_CFG_UNSIGNED, &cfg.maxNodeRequests    },
      { "minServiceSlots",    DNX_CFG_UNSIGNED, &cfg.minServiceSlots    },
      { "expirePollInterval", DNX_CFG_UNSIGNED, &cfg.expirePollInterval },
      { "localCheckPattern",  DNX_CFG_STRING,   &cfg.localCheckPattern  },
      { "syncScript",         DNX_CFG_FSPATH,   &cfg.syncScript         },
      { "logFile",            DNX_CFG_FSPATH,   &cfg.logFilePath        },
      { "debugFile",          DNX_CFG_FSPATH,   &cfg.debugFilePath      },
      { "auditFile",          DNX_CFG_FSPATH,   &cfg.auditFilePath      },
      { "debugLevel",         DNX_CFG_UNSIGNED, &cfg.debugLevel         },
      { 0 },
   };
   char cfgdefs[] =
      "channelAgent = udp://0:12482\n"
      "channelDispatcher = udp://0:12480\n"
      "channelCollector = udp://0:12481\n"
      "maxNodeRequests = 0x7FFFFFFF\n"
      "minServiceSlots = 100\n"
      "expirePollInterval = 5\n"
      "logFile = " DNX_DEFAULT_LOG "\n"
      "debugFile = " 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 * escaped_res_data;
   char * filename;
   FILE * fp = 0;
   int fd;

   // a nagios 3.x global variable
   extern char * temp_path;

   // a nagios 3.x core function
   if ((escaped_res_data = escape_newlines(res_data)) == 0)
      return DNX_ERR_MEMORY;

   // 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)
   {
      free(escaped_res_data); // allocated by nagios - use free
      return DNX_ERR_MEMORY;
   }
   sprintf(filename, "%s/checkXXXXXX", temp_path);
   if ((fd = mkstemp(filename)) < 0
         || (fp = fdopen(fd, "w")) == 0)
   {
      xfree(filename);
      free(escaped_res_data); // allocated by nagios - use free
      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", escaped_res_data);

   fclose(fp);

   free(escaped_res_data); // allocated by nagios - use free

   nagios3xMoveCheckResultToQueue(filename);

   xfree(filename);

   return 0;
}

#endif // ?DIRECT_POST

#else
# error Unsupported NEB API version.
#endif

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

int dnxPostResult(void * data, time_t start_time, unsigned delta,
      int early_timeout, int res_code, char * res_data)
{
   DnxJobData * jdp = (DnxJobData *)data;
   int ret;

   if (early_timeout)
      res_code = STATE_UNKNOWN;

   // update stats
   if (early_timeout)
      dnxStatsInc(0, RESULTS_TIMED_OUT);
   else if (!res_code)
      dnxStatsInc(0, RESULTS_OK);
   else
      dnxStatsInc(0, RESULTS_FAILED);

#if CURRENT_NEB_API_VERSION == 2

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

#elif CURRENT_NEB_API_VERSION == 3

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

#else
# error Unsupported NEB API version.
#endif

   // update stats
   if (ret == 0)
      dnxStatsInc(0, POST_RESULTS_OK);
   else
      dnxStatsInc(0, POST_RESULTS_FAILED);

   return ret;
}

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

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);
   }

   // check for configuration minServiceSlots override
   if (size < cfg.minServiceSlots)
   {
      dnxLog("Overriding calculated service check slot count. "
             "Increasing from %d to configured minimum: %d.",
             size, cfg.minServiceSlots);
      size = cfg.minServiceSlots;
   }

   // check for configuration maxNodeRequests override
   if (size > cfg.maxNodeRequests)
   {
      dnxLog("Overriding calculated service check slot count. "
             "Decreasing from %d to configured maximum: %d.", size,
             cfg.maxNodeRequests);
      size = cfg.maxNodeRequests;
   }
   return size;
}

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

static int dnxPostNewJob(DnxJobList * joblist, unsigned long serial, DnxJobData * jdp, nebstruct_service_check_data * ds, DnxNodeRequest * pNode)
{
   DnxNewJob Job;
   int ret;

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

   // fill-in the job structure with the necessary information
   dnxMakeXID(&Job.xid, DNX_OBJ_JOB, serial, 0);
   Job.payload    = jdp;
   Job.cmd        = xstrdup(ds->command_line);
   Job.start_time = ds->start_time.tv_sec;
   Job.timeout    = ds->timeout;
   Job.expires    = Job.start_time + Job.timeout + 5;
   Job.pNode      = pNode;

   dnxDebug(2, "DnxNebMain: Posting Job [%lu]: %s.", serial, Job.cmd);

   // post to the Job Queue
   if ((ret = dnxJobListAdd(joblist, &Job)) != DNX_OK)
   {
      dnxStatsInc(0, JOBS_REJECTED_NO_SLOTS);
      dnxLog("Failed to post Job [%lu]; \"%s\": %d.",
            Job.xid.objSerial, Job.cmd, ret);
   }
   else
   {
      dnxStatsInc(0, JOBS_HANDLED);
      dnxAuditJob(&Job, "ASSIGN");
   }
   return ret;
}

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

static int ehSvcCheck(int event_type, void * data)
{
   static unsigned long serial = 0; // the number of service checks processed

   nebstruct_service_check_data * svcdata = (nebstruct_service_check_data *)data;
   DnxNodeRequest * pNode;
   DnxJobData * jdp;
   int ret;

   if (event_type != NEBCALLBACK_SERVICE_CHECK_DATA)
      return OK;

   if (svcdata == 0)
   {
      dnxLog("Service handler received NULL service data structure.");
      return ERROR;  // shouldn't happen - internal Nagios error
   }

   if (svcdata->type != NEBTYPE_SERVICECHECK_INITIATE)
      return OK;  // ignore non-initiate service checks

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

   dnxDebug(3, "ehSvcCheck: Received Job [%lu] at %lu (%lu).",
         serial, (unsigned long)time(0),
         (unsigned long)svcdata->start_time.tv_sec);

   if ((ret = dnxGetNodeRequest(registrar, &pNode)) != DNX_OK)
   {
      dnxDebug(3, "ehSvcCheck: No worker nodes requests available: %s.",dnxErrorString(ret));
      return OK;     // tell nagios execute locally
   }

   // allocate and populate a new job payload object
   if ((jdp = (DnxJobData *)xmalloc(sizeof *jdp)) == 0)
   {
      dnxDebug(1, "ehSvcCheck: Out of memory!");
      return OK;
   }
   memset(jdp, 0, sizeof *jdp);
   jdp->svc = (service *)svcdata->OBJECT_FIELD_NAME;

   assert(jdp->svc);

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

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

   if ((ret = dnxPostNewJob(joblist, serial, jdp, svcdata, pNode)) != DNX_OK)
   {
      dnxLog("Unable to post job [%lu]: %s.", serial, dnxErrorString(ret));
      xfree(jdp);
      return OK;     // tell nagios execute locally
   }

   serial++;                           // bump serial number
   return NEBERROR_CALLBACKOVERRIDE;   // tell nagios we want it
}

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

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

static int dnxServerDeInit(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

   dnxReleaseAgent();

   // ensure we don't destroy non-existent objects from here on out...
   if (registrar)
      dnxRegistrarDestroy(registrar);

   if (dispatcher)
      dnxDispatcherDestroy(dispatcher);

   if (collector)
      dnxCollectorDestroy(collector);

   if (joblist)
      dnxJobListDestroy(joblist);

   // it doesn't matter if we haven't initialized the
   // channel map - it can figure that out for itself
   dnxChanMapRelease();

   dnxStatsCleanup();

   return OK;
}

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

static int dnxServerInit(void)
{
   int ret, joblistsz;

   // clear globals so we know what to "undo" as we back out
   joblist = 0;
   registrar = 0;
   dispatcher = 0;
   collector = 0;

   if ((ret = dnxChanMapInit(0)) != 0)
   {
      dnxLog("Failed to initialize channel map: %s.", dnxErrorString(ret));
      return ret;
   }

   joblistsz = dnxCalculateJobListSize();

   dnxLog("Allocating %d service request slots in the DNX job list.", joblistsz);

   if ((ret = dnxJobListCreate(joblistsz, &joblist)) != 0)
   {
      dnxLog("Failed to initialize DNX job list with %d slots.", joblistsz);
      return ret;
   }

   // create and configure collector
   if ((ret = dnxCollectorCreate("Collect", cfg.collectorUrl,
               joblist, &collector)) != 0)
      return ret;

   // create and configure dispatcher
   if ((ret = dnxDispatcherCreate("Dispatch", cfg.dispatcherUrl,
               joblist, &dispatcher)) != 0)
      return ret;

   // create worker node registrar
   if ((ret = dnxRegistrarCreate(joblistsz * 2,
               dnxDispatcherGetChannel(dispatcher), &registrar)) != 0)
      return ret;

   // initialize server management agent
   if ((ret = dnxInitAgent(cfg.agentUrl, parser)) != 0)
      return ret;

#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("Server initialization completed.");

   return 0;
}

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

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
   assert(procdata);
   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 server init fails, do server shutdown
      if (dnxServerInit() != 0)
         dnxServerDeInit();
   }
   return OK;
}

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

char * versionText(void)
{
   static char versionText[] = 
      "\n"
      "  dnxServer Version " VERSION ", Built " __DATE__ " at " __TIME__ ".\n"
      "  Distributed Nagios eXecutor (DNX) Integrated Plugin/Server Library.\n"
      "  Please report bugs to <" PACKAGE_BUGREPORT ">.\n"
      "\n"
      "  Default configuration:\n"
      "    Default config file: "      DNX_DEFAULT_SERVER_CONFIG_FILE "\n"
      "    Default log file: "         DNX_DEFAULT_LOG "\n"
      "    Default debug log file: "   DNX_DEFAULT_DBGLOG "\n"
//    "    Compile flags: "            COMPILE_FLAGS "\n"
#if DEBUG_HEAP
      "    Debug heap is ENABLED.\n"
#endif
#if DEBUG_LOCKS
      "    Debug locks are ENABLED.\n"
#endif
      ;
   return xstrdup(versionText);
}

void dnxJobCleanup(DnxNewJob * pJob)
{
   if (pJob)
   {
      xfree(pJob->cmd);
      xfree(pJob->payload);
      xfree(pJob->pNode);
   }
}

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

int dnxAuditJob(DnxNewJob * pJob, char * action)
{
   if (cfg.auditFilePath)
      return dnxAudit(
            "%s: Job %lu: Worker %s-%lx: %s",
                  action, pJob->xid.objSerial,
                  pJob->pNode->addrstr, 
                  pJob->pNode->xid.objSlot, pJob->cmd);

   return DNX_OK;
}

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

int nebmodule_deinit(int flags, int reason)
{
   dnxLog("-------- DNX Server Module Shutdown Initiated --------");
   dnxServerDeInit();

   releaseConfig();
   xheapchk();

   dnxLog("-------- DNX Server 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;

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

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

   dnxLog("-------- DNX Server Module Version %s Startup --------", VERSION);
   dnxLog("Copyright (c) 2006-2010 Intellectual Reserve. All rights reserved.");
   dnxLog("Configuration file: %s.", args);
   dnxLog("Dispatcher: %s.", cfg.dispatcherUrl);
   dnxLog("Collector: %s.", cfg.collectorUrl);
   dnxLog("Agent: %s.", cfg.agentUrl);
   if (cfg.debugFilePath && cfg.debugLevel != 0)
   {
      dnxLog("Debug logging enabled at level %d to %s.",
            cfg.debugLevel, cfg.debugFilePath);
#if DEBUG_HEAP
      dnxLog("Debug heap is enabled.");
#endif
#if DEBUG_LOCKS
      dnxLog("Debug locks are enabled.");
#endif
   }
   if (cfg.auditFilePath)
      dnxLog("Auditing enabled to %s.", cfg.auditFilePath);

#if DEBUG_HEAP
      dnxLog("Debug heap is enabled.");
#endif
#if DEBUG_LOCKS
      dnxLog("Debug locks are enabled.");
#endif

   // 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));
      releaseConfig();
      return ERROR;
   }
   start_time = time(0);

   dnxLog("-------- DNX Server Module Startup Complete --------");

   return OK;
}