dnxServerMain.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"
#else
# define VERSION "<unknown>"
#endif

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

#if HAVE_GETOPT_LONG
#define _GNU_SOURCE
#include <getopt.h>
#endif

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

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

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

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <stdarg.h>
#include <signal.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>

#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_PATH        SYSBINDIR  "/dnxServer"

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

// module static data
static DnxCfgData cfg;              
static DnxCfgParser * parser;       
static DnxJobList * joblist;        
static DnxRegistrar * registrar;    
static DnxDispatcher * dispatcher;  
static DnxCollector * collector;    
static time_t start_time;           
static char * s_cfgfile;            
static int s_queuesize;             
static char * s_progname;           
static char * s_cmdover = 0;        

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

static void version(FILE * fp)
{
   char * vertxt = versionText();
   if (vertxt)
   {
      fprintf(fp, "%s\n", vertxt);
      xfree(vertxt);
      vertxt = 0;
   }
}

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

static void usage(void)
{

#if HAVE_GETOPT_LONG
# define OL_CFGFILE   ", --cfgfile  "
# define OL_QUEUESIZE ", --queuesize"
# define OL_VERSION   ", --version  "
# define OL_HELP      ", --help     "
#else
# define OL_CFGFILE
# define OL_QUEUESIZE
# define OL_VERSION
# define OL_HELP
#endif

   version(stderr);
   fprintf(stderr,
      "  Usage: %s [options]\n"
      "    Where [options] are:\n"
      "      -c" OL_CFGFILE   " <file>   specify the file and path of the config file.\n"
      "      -j" OL_QUEUESIZE " <num>    specify the size of the Nagios service queue.\n"
      "      -v" OL_VERSION   "          display DNX server version and exit.\n"
      "      -h" OL_HELP      "          display this help screen and exit.\n"
      "\n",
      s_progname
   );
   exit(-1);
}

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

static int getOptions(int argc, char ** argv)
{
// extern int optind;
   extern char * optarg;
   extern int opterr, optopt;

   static char opts[] = "c:j:vh";

#if HAVE_GETOPT_LONG
   static struct option longopts[] =
   {
      { "cfgfile",  required_argument, 0, 'c' },
      { "queuesize",required_argument, 0, 'j' },
      { "version",  no_argument,       0, 'v' },
      { "help",     no_argument,       0, 'h' },
      { 0, 0, 0, 0 },
   };
#endif

   int ch;
   char * cp;

   // set program base name
   s_progname = (char *)((cp = strrchr(argv[0], '/')) != 0 ? (cp + 1) : argv[0]);

   opterr = 0; /* Disable error messages */

#if HAVE_GETOPT_LONG
   while ((ch = getopt_long(argc, argv, opts, longopts, 0)) != -1)
#else
   while ((ch = getopt(argc, argv, opts)) != -1)
#endif
   {
      switch (ch)
      {
         case 'c': s_cfgfile = optarg; break;
         case 'j': s_queuesize = atoi(optarg); break;
         case 'v': version(stdout); exit(0);
         case 'h':
         default : usage();
      }
   }

   if (!s_cfgfile)
      s_cfgfile = DNX_DEFAULT_SERVER_CONFIG_FILE;

   return 0;
}

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

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

   assert(dict && vptrs);

   // setup data structure so we can use the same functionality we had before
   cfg.dispatcherUrl      = (char *)vptrs[ 0];
   cfg.collectorUrl       = (char *)vptrs[ 1];
   cfg.agentUrl           = (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.logFilePath        = (char *)vptrs[ 7];
   cfg.debugFilePath      = (char *)vptrs[ 8];
   cfg.debugLevel         = (unsigned)(intptr_t)vptrs[ 9];
   cfg.auditFilePath      = (char *)vptrs[10];

   // validate configuration items in context
   if (!cfg.dispatcherUrl)
      dnxLog("config: Missing channelDispatcher parameter.");
   else if (!cfg.collectorUrl)
      dnxLog("config: Missing channelCollector parameter.");
   else if (!cfg.agentUrl)
      dnxLog("config: Missing channelAgent 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
      ret = 0;

   return ret;
}

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

static int initConfig(char * cfgfile)
{
   DnxCfgDict dict[] =
   {  // Do NOT change the order, unless you know what you're doing!
      { "channelDispatcher",  DNX_CFG_URL,      &cfg.dispatcherUrl      },
      { "channelCollector",   DNX_CFG_URL,      &cfg.collectorUrl       },
      { "channelAgent",       DNX_CFG_URL,      &cfg.agentUrl           },
      { "authWorkerNodes",    DNX_CFG_STRING,   &cfg.authWorkerNodes    },
      { "maxNodeRequests",    DNX_CFG_UNSIGNED, &cfg.maxNodeRequests    },
      { "minServiceSlots",    DNX_CFG_UNSIGNED, &cfg.minServiceSlots    },
      { "expirePollInterval", DNX_CFG_UNSIGNED, &cfg.expirePollInterval },
      { "logFile",            DNX_CFG_FSPATH,   &cfg.logFilePath        },
      { "debugFile",          DNX_CFG_FSPATH,   &cfg.debugFilePath      },
      { "debugLevel",         DNX_CFG_UNSIGNED, &cfg.debugLevel         },
      { "auditFile",          DNX_CFG_FSPATH,   &cfg.auditFilePath      },
      { 0 },
   };
   char cfgdefs[] =
      "channelDispatcher = udp://0:12480\n"
      "channelCollector = udp://0:12481\n"
      "channelAgent = udp://0:12482\n"
      "maxNodeRequests = 0x7FFFFFFF\n"
      "minServiceSlots = 100\n"
      "expirePollInterval = 5\n"
      "logFile = " DNX_DEFAULT_LOG "\n"
      "debugFile = " DNX_DEFAULT_DBGLOG "\n";

   int ret;

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

   // parse configuration file; pass defaults
   if ((ret = dnxCfgParserParse(parser, 0)) != 0)
      dnxCfgParserDestroy(parser);

   return ret;
}

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

static void freeCfgData(DnxCfgData * cpy)
{
   xfree(cpy->dispatcherUrl);
   xfree(cpy->collectorUrl);
   xfree(cpy->agentUrl);
   xfree(cpy->authWorkerNodes);
   xfree(cpy->logFilePath);
   xfree(cpy->debugFilePath);
   xfree(cpy->auditFilePath);
   xfree(cpy);
   cpy = 0;
}

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

static DnxCfgData * copyCfgData(DnxCfgData * org)
{
   DnxCfgData * cpy;

   // make new config structure
   if ((cpy = (DnxCfgData *)xmalloc(sizeof *cpy)) == 0)
      return 0;

   // copy all values
   *cpy = *org;

   // attempt to make string buffer copies
   cpy->dispatcherUrl = xstrdup(org->dispatcherUrl);
   cpy->collectorUrl = xstrdup(org->collectorUrl);
   cpy->agentUrl = xstrdup(org->agentUrl);
   cpy->authWorkerNodes = xstrdup(org->authWorkerNodes);
   cpy->maxNodeRequests = org->maxNodeRequests;
   cpy->minServiceSlots = org->minServiceSlots;
   cpy->expirePollInterval = org->expirePollInterval;
   cpy->logFilePath = xstrdup(org->logFilePath);
   cpy->debugFilePath = xstrdup(org->debugFilePath);
   cpy->debugLevel = org->debugLevel;
   cpy->auditFilePath = xstrdup(org->auditFilePath);

   return cpy;
}

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

static void logGblConfigChanges(DnxCfgData * ocp, DnxCfgData * ncp)
{
   if (strcmp(ocp->dispatcherUrl, ncp->dispatcherUrl) != 0)
      dnxLog("Config parameter 'channelDispatcher' changed from %s to %s. "
            "NOTE: Changing the Dispatcher URL requires a restart.",
            ocp? ocp->dispatcherUrl: "<unknown>", ncp->dispatcherUrl);

   if (strcmp(ocp->collectorUrl, ncp->collectorUrl) != 0)
      dnxLog("Config parameter 'channelCollector' changed from %s to %s. "
            "NOTE: Changing the Collector URL requires a restart.",
            ocp? ocp->collectorUrl: "<unknown>", ncp->collectorUrl);

   if (strcmp(ocp->agentUrl, ncp->agentUrl) != 0)
      dnxLog("Config parameter 'channelAgent' changed from %s to %s. "
            "NOTE: Changing the Agent URL requires a restart.",
            ocp? ocp->agentUrl: "<unknown>", ncp->agentUrl);

   if (strcmp(ocp->authWorkerNodes, ncp->authWorkerNodes) != 0)
      dnxLog("Config parameter 'authWorkerNodes' changed from %s to %s. ",
            ocp? ocp->authWorkerNodes: "<unknown>", ncp->authWorkerNodes);

   if (ocp->maxNodeRequests != ncp->maxNodeRequests)
      dnxLog("Config parameter 'maxNodeRequests' changed from %u to %u. "
            "NOTE: Changing the maximum node requests requires a restart.",
            ocp? ocp->maxNodeRequests: 0, ncp->maxNodeRequests);

   if (ocp->minServiceSlots != ncp->minServiceSlots)
      dnxLog("Config parameter 'minServiceSlots' changed from %u to %u. "
            "NOTE: Changing the minimum service requests requires a restart.",
            ocp? ocp->minServiceSlots: 0, ncp->minServiceSlots);

   if (ocp->expirePollInterval != ncp->expirePollInterval)
      dnxLog("Config parameter 'expirePollInterval' changed from %u to %u. ",
            ocp? ocp->expirePollInterval: 0, ncp->expirePollInterval);

   if (strcmp(ocp->logFilePath, ncp->logFilePath) != 0)
      dnxLog("Config parameter 'logFile' changed from %s to %s. "
            "NOTE: Changing the server log file path requires a restart.",
            ocp? ocp->logFilePath: "<unknown>", ncp->logFilePath);

   if (strcmp(ocp->debugFilePath, ncp->debugFilePath) != 0)
      dnxLog("Config parameter 'debugFile' changed from %s to %s. "
            "NOTE: Changing the server debug log file path requires a restart.",
            ocp? ocp->debugFilePath: "<unknown>", ncp->debugFilePath);

   if (ocp->debugLevel != ncp->debugLevel)
      dnxLog("Config parameter 'debugLevel' changed from %u to %u.",
            ocp? ocp->debugLevel: 0, ncp->debugLevel);
   
   if (strcmp(ocp->auditFilePath, ncp->auditFilePath) != 0)
      dnxLog("Config parameter 'auditFile' changed from %s to %s. "
            "NOTE: Changing the audit log file path requires a restart.",
            ocp? ocp->auditFilePath: "<unknown>", ncp->auditFilePath);
}

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

static int dnxCalculateJobListSize(void)
{
   int size = s_queuesize; // get value passed (maybe) on the cmdline

   // 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 requested 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 requested service check slot count. "
             "Decreasing from %d to configured maximum: %d.", size,
             cfg.maxNodeRequests);
      size = cfg.maxNodeRequests;
   }
   return size;
}

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

static int dnxPostNewJob(DnxJobList * joblist, DnxJobTransfer * job,
      DnxNodeRequest * pNode)
{
   int ret;
   DnxNewJob Job;

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

   dnxDebug(2, "Posting New Job [%lu]: %s.", job->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 ProcessJobTransferMsg(DnxJobTransfer * job, DnxNodeRequest * node)
{
   size_t toread = job->hdr.structsz - sizeof job->hdr;
   char * bufptr = (char *)job + sizeof job->hdr;
   ssize_t bytesread;
   int ret;

   // read the rest of the job transfer structure
   while (toread > 0 && (bytesread = read(DNX_SRVFD, 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, "ProcessJobTransferMsg: read failed: %s.", strerror(ret));
      return ret;
   }

   // process service check request
   if (node || (ret = dnxGetNodeRequest(registrar, &node)) == 0)
      ret = dnxPostNewJob(joblist, job, node);

   if (ret != 0)
      dnxDebug(1, "ProcessJobTransferMsg: %s.", dnxErrorString(ret));

   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(DNX_SRVFD, bufptr, toread)) > 0)
   {
      toread -= bytesread;
      bufptr += bytesread;
   }

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

   // check for error
   if (bytesread < 0)
   {
      int ret = bytesread == 0? EPIPE: 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 SendMsgResponse(int result)
{
   ssize_t byteswritten;
   size_t towrite = sizeof result;
   char * bufptr = (char *)&result;

   // write ack/nak to plugin
   while (towrite > 0 && (byteswritten = write(DNX_SAKFD, bufptr, towrite)) > 0)
   {
      towrite -= byteswritten;
      bufptr += byteswritten;
   }

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

   // check for error
   if (byteswritten < 0)
   {
      int ret = byteswritten == 0? EPIPE: errno;
      dnxDebug(1, "SendMsgResponse: write failed: %s.", strerror(ret));
      return ret;
   }

   return 0;
}

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

static int processRequests()
{
   DnxNodeRequest * node = 0;
   DnxJobTransfer * job = 0;
   size_t jobsz = 0;
   int terminating = 0;

   dnxLog("DNX Server awaiting service check requests...");

   while (!terminating)
   {
      int ret;
      DnxMsgHeader hdr;
      if ((ret = GetMsgHeader(&hdr)) == 0)
      {
         switch (hdr.msgtype)
         {
            case DNX_MSG_RESERVE_NODEREQ:
               ret = dnxGetNodeRequest(registrar, &node);
               break;

            case DNX_MSG_JOB_TRANSFER:
               // resize job transfer buffer if necessary
               if (hdr.structsz > jobsz)
               {
                  DnxJobTransfer * tmp;
                  if ((tmp = (DnxJobTransfer *)xrealloc(job, hdr.structsz)) != 0)
                  {
                     job = tmp;
                     jobsz = hdr.structsz;
                  }
               }

               // copy header and read rest of job transfer message
               job->hdr = hdr;
               if ((ret = ProcessJobTransferMsg(job, node)) == 0)
                  node = 0;
               break;

            case DNX_MSG_TERMINATE:
               terminating = 1;
               break;
         }
         SendMsgResponse(ret);
      }
      if (ret == EPIPE)
         terminating = 1;
   }

   dnxDebug(2, "Request loop terminating.");

   close(DNX_SRVFD);
   close(DNX_SAKFD);

   xfree(job);
   xfree(node);
   return 0;
}

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

static int dnxServerDeInit(void)
{
   // ensure we don't attempt to destroy non-existent objects

   dnxReleaseAgent();

   if (registrar)
      dnxRegistrarDestroy(registrar);

   if (collector)
      dnxCollectorDestroy(collector);

   if (dispatcher)
      dnxDispatcherDestroy(dispatcher);

   if (joblist)
      dnxJobListDestroy(joblist);

   dnxChanMapRelease();
   dnxStatsCleanup();

   return 0;
}

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

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;

   // initialize the channel map
   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: %s.",
            joblistsz, dnxErrorString(ret));
      return ret;
   }

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

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

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

   // initialize server agent comm and thread
   if ((ret = dnxInitAgent(cfg.agentUrl)) != 0)
      return ret;

   dnxLog("Server initialization completed.");

   return 0;
}

static void sighandler(int sig)
{
   switch(sig)
   {
      case SIGHUP: dnxAgentSignalReconfig(); break;
      break;
   }
}

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

char * versionText(void)
{
   char buf[1024];
   snprintf(buf, sizeof(buf) - 1,
      "\n"
      "  %s Version " VERSION ", Built " __DATE__ " at " __TIME__ ".\n"
      "  Distributed Nagios eXecutor (DNX) Server Daemon.\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
      , s_progname
   );
   return xstrdup(buf);
}

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

void dnxReconfigure(void)
{
   int ret;
   DnxCfgData * old;

   dnxLog("RECONFIGURE request received. Reconfiguring...");

   // reparse config file into temporary cfg structure and validate
   old = copyCfgData(&cfg);
   if ((ret = dnxCfgParserParse(parser, 0)) == 0)
      logGblConfigChanges(old, &cfg);
   if (old) freeCfgData(old);

   dnxLog("Reconfiguration: %s.", dnxErrorString(ret));
}

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

int dnxPostResult(void * payload, unsigned long serial, time_t start_time, 
      unsigned delta, int early_timeout, int res_code, char * res_data)
{
   int ret = 0;
   DnxResultTransfer * result;
   ssize_t byteswritten;
   char * bufptr;
   size_t resultsz;

   dnxDebug(2, "Transferring result [%lu] to plugin: %s.", serial, res_data);

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

   // allocate a results transfer buffer
   resultsz = sizeof(*result) + strlen(res_data);
   if ((result = (DnxResultTransfer *)xmalloc(resultsz)) == 0)
      return DNX_ERR_MEMORY;

   // fill in result transfer structure
   result->hdr.structsz = resultsz;
   result->hdr.signature = DNX_MSG_SIGNATURE;
   result->hdr.msgtype = DNX_MSG_RESULT_TRANSFER;
   result->serial = serial;
   result->payload = payload;
   result->start_time = start_time;
   result->delta = delta;
   result->early_timeout = early_timeout;
   result->res_code = res_code;
   strcpy(result->res_data, res_data);

   // write result to server
   bufptr = (char *)result;
   while (resultsz > 0 && (byteswritten = write(DNX_PLGFD, bufptr, resultsz)) > 0)
   {
      bufptr += byteswritten;
      resultsz -= byteswritten;
   }

   // check for error or broken pipe
   if (byteswritten <= 0)
      ret = byteswritten == 0? EPIPE: errno;

   xfree(result);

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

   return ret;
}

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

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

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

int dnxAuditJob(DnxNewJob * pJob, char * action)
{
   if (cfg.auditFilePath)
   {
      struct sockaddr_in src_addr;
      in_addr_t addr;

      // Convert opaque Worker Node address to IPv4 address

      memcpy(&src_addr, pJob->pNode->address, sizeof(src_addr));
      addr = ntohl(src_addr.sin_addr.s_addr);

      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 main(int argc, char ** argv)
{
   int ret;

   // parse command line options; read configuration file
   if ((ret = getOptions(argc, argv)) != DNX_OK
         || (ret = initConfig(s_cfgfile)) != DNX_OK)
      goto e0;

   // initialize the logging subsystem with configured settings
   dnxLogInit(cfg.logFilePath, cfg.debugFilePath, cfg.auditFilePath, 
         &cfg.debugLevel);

   dnxLog("-------- DNX Server Version %s Startup --------", VERSION);
   dnxLog("Copyright (c) 2006-2010 Intellectual Reserve. All rights reserved.");
   dnxLog("Configuration file: %s.", s_cfgfile);
   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);

   start_time = time(0);

   // install signal handlers
   signal(SIGHUP,  sighandler);
   signal(SIGINT,  sighandler);
   signal(SIGQUIT, sighandler);
   signal(SIGABRT, sighandler);
   signal(SIGTERM, sighandler);
   signal(SIGPIPE, SIG_IGN);
   signal(SIGALRM, SIG_IGN);
   signal(SIGUSR1, sighandler);
   signal(SIGUSR2, SIG_IGN);

   if ((ret = dnxServerInit()) != 0)
      goto e2;

   //----------------------------------------------------------------------
   ret = processRequests();
   //----------------------------------------------------------------------

   dnxDebug(1, "Command-loop exited: %s.", dnxErrorString(ret));

e2:dnxServerDeInit();
e1:dnxCfgParserDestroy(parser);
e0:dnxLog("-------- DNX Server Daemon Shutdown Complete --------");

   xheapchk();
   closelog();

   return ret;
}

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