dnxAgent.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 "dnxAgent.h"
#include "dnxServerMain.h"
#include "dnxTransport.h"
#include "dnxProtocol.h"
#include "dnxCfgParser.h"
#include "dnxError.h"
#include "dnxDebug.h"
#include "dnxLogging.h"
#include "dnxStats.h"

#include <unistd.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>

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

// module static data
static char * s_agentName = "Agent";   
static DnxChannel * s_agent;           
static pthread_t s_agentTid;           
static DnxCfgParser * s_parser;        
static int s_shutdown = 0;             
static int s_reconfig = 0;             

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

static char * buildHelpReply(void)
{
   static char * help =
         "DNX Server Management Commands:\n"
         "  SHUTDOWN\n"
         "  RECONFIGURE\n"
         "  RESETSTATS\n"
         "  GETSTATS stat-list\n"
         "  GETNODESTATS ip-address stat-list\n"
         "    ip-address is a dot-delimited IPv4 node address or DNS name.\n"
         "    stat-list is a comma-delimited list of stat names:\n"
         "      Server only:\n"
         "        nodes_registered       - total client nodes registered\n"
         "        jobs_handled           - jobs accepted from Nagios\n"
         "        jobs_rejected_no_slots - jobs rejected due to no job list slots\n"
         "        jobs_rejected_no_nodes - jobs rejected due to no node requests\n"
         "        post_results_ok        - results successfully posted to Nagios\n"
         "        post_results_failed    - results failing post to Nagios\n"
         "      Server and individual nodes:\n"
         "        requests_received      - total requests received from all nodes\n"
         "        requests_expired       - total requests that have expired\n"
         "        dispatches_ok          - jobs successfully dispatched to nodes\n"
         "        dispatches_failed      - jobs failing dispatch to nodes\n"
         "        results_ok             - results returned with zero result code\n"
         "        results_failed         - results returned with non-zero result code\n"
         "        results_timed_out      - results timed out from assigned node\n"
         "    Note: Stats are returned in the order they are requested.\n"
         "          Unrecognized requested stats are returned as '?'.\n"
         "  GETNODELIST\n"
         "  GETCONFIG\n"
         "  GETVERSION\n"
         "  HELP";
   return xstrdup(help);
}

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

static int appendString(char ** spp, char * fmt, ... )
{
   char buf[1024];
   char * newstr;
   size_t strsz;
   va_list ap;

   // build new string
   va_start(ap, fmt);
   vsnprintf(buf, sizeof buf, fmt, ap);
   va_end(ap);

   // reallocate buffer; initialize if necessary
   strsz = strlen(buf) + 1;
   if ((newstr = xrealloc(*spp, (*spp? strlen(*spp): 0) + strsz)) == 0) 
      return DNX_ERR_MEMORY;
   if (*spp == 0)
      *newstr = 0;

   // concatenate new string onto exiting string; return updated pointer
   strcat(newstr, buf);
   *spp = newstr;
   return 0;
}

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

static int dnxCountNodes(DnxNodeData * node, void * data)
{
   // only count active nodes
   if (node->stats[REQUESTS_RECEIVED] != 0)
      (*(int *)data)++;
   return 0;
}

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

static char * buildMgmtStatsReply(char * req)
{
   char * rsp = 0;
   int nodes_registered = 0;
   unsigned stats[STATS_COUNT];

   struct { char * str; unsigned * stat; } rs[] =
   {
      { "nodes_registered",       &nodes_registered              },
      { "requests_received",      &stats[REQUESTS_RECEIVED]      },
      { "requests_expired",       &stats[REQUESTS_EXPIRED]       },
      { "dispatches_ok",          &stats[DISPATCHES_OK]          },
      { "dispatches_failed",      &stats[DISPATCHES_FAILED]      },
      { "results_ok",             &stats[RESULTS_OK]             },
      { "results_failed",         &stats[RESULTS_FAILED]         },
      { "results_timed_out",      &stats[RESULTS_TIMED_OUT]      },
      { "jobs_handled",           &stats[JOBS_HANDLED]           },
      { "jobs_rejected_no_slots", &stats[JOBS_REJECTED_NO_SLOTS] },
      { "jobs_rejected_no_nodes", &stats[JOBS_REJECTED_NO_NODES] },
      { "post_results_ok",        &stats[POST_RESULTS_OK]        },
      { "post_results_failed",    &stats[POST_RESULTS_FAILED]    },
   };

   assert(req);

   // count number of registered nodes
   dnxStatsForEachNode(dnxCountNodes, &nodes_registered);

   // get a copy of the server stats
   dnxStatsGetServerStats(stats);

   // trim leading ws
   while (isspace(*req)) req++;

   while (*req)
   {
      char * ep, * np;
      unsigned i;

      // find start of next string or end
      if ((np = strchr(req, ',')) == 0)
         np = req + strlen(req);

      // trim trailing ws
      ep = np;
      while (ep > req && isspace(ep[-1])) ep--;

      // search table for sub-string, append requested stat to rsp
      for (i = 0; i < elemcount(rs); i++)
         if (memcmp(req, rs[i].str, ep - req) == 0)
         {
            if (appendString(&rsp, "%u,", *rs[i].stat) != 0)
               return xfree(rsp), (char *)0;
            break;
         }

      // check for unknown stat
      if (i == elemcount(rs) && appendString(&rsp, "?,") != 0)
         return xfree(rsp), (char *)0;

      // move to next sub-string or end
      if (*(req = np)) req++;

      // trim leading ws
      while (isspace(*req)) req++;
   }

   // remove trailing comma in non-empty response
   if (rsp)
   {
      size_t len = strlen(rsp);
      if (len && rsp[len - 1] == ',') rsp[len - 1] = 0;
   }
   return rsp;
}

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

typedef struct MgmtNodeStatsData
{
   char * stats;
   char * rsp;
} MgmtNodeStatsData;

static int dnxBuildNodeStatsReply(DnxNodeData * node, void * data)
{
   MgmtNodeStatsData * msd = (MgmtNodeStatsData *)data;
   char * req = msd->stats;
   char * rsp = 0;

   struct { char * str; unsigned * stat; } rs[] =
   {
      { "requests_received",  &node->stats[REQUESTS_RECEIVED] },
      { "requests_expired",   &node->stats[REQUESTS_EXPIRED]  },
      { "dispatches_ok",      &node->stats[DISPATCHES_OK]     },
      { "dispatches_failed",  &node->stats[DISPATCHES_FAILED] },
      { "results_ok",         &node->stats[RESULTS_OK]        },
      { "results_failed",     &node->stats[RESULTS_FAILED]    },
      { "results_timed_out",  &node->stats[RESULTS_TIMED_OUT] },
   };

   msd->rsp = 0;

   // check for inactive node status - return empty if inactive
   if (node->stats[REQUESTS_RECEIVED] == 0)
      return DNX_OK;

   while (*req)
   {
      char * ep, * np;
      unsigned i;

      // find start of next string or end
      if ((np = strchr(req, ',')) == 0)
         np = req + strlen(req);

      // trim trailing ws
      ep = np;
      while (ep > req && isspace(ep[-1])) ep--;

      // search table for sub-string, append requested stat to rsp
      for (i = 0; i < elemcount(rs); i++)
         if (memcmp(req, rs[i].str, ep - req) == 0)
         {
            if (appendString(&rsp, "%u,", *rs[i].stat) != 0)
               return xfree(rsp), DNX_ERR_MEMORY;
            break;
         }

      // check for unknown stat
      if (i == elemcount(rs) && appendString(&rsp, "?,") != 0)
         return xfree(rsp), DNX_ERR_MEMORY;

      // move to next sub-string or end
      if (*(req = np)) req++;

      // trim leading ws
      while (isspace(*req)) req++;
   }

   // remove trailing comma in non-empty response
   if (rsp)
   {
      size_t len = strlen(rsp);
      if (len && rsp[len - 1] == ',') rsp[len - 1] = 0;
   }

   msd->rsp = rsp;
   return DNX_OK;
}

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

static char * buildMgmtNodeStatsReply(char * req)
{
   MgmtNodeStatsData data;

   assert(req);

   // req points to ip-address, remove leading ws
   while (isspace(*req)) req++;

   // locate stats-list, error if missing
   if ((data.stats = strchr(req, ' ')) == 0)
      return xstrdup("");

   // data.stats points to stats-list, remove leading ws
   *data.stats++ = 0;
   while (isspace(*data.stats)) data.stats++;

   dnxStatsForNodeByAddrStr(req, dnxBuildNodeStatsReply, &data);

   return data.rsp? data.rsp: xstrdup("");
}

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

static int dnxResetNodeStats(DnxNodeData * node, void * data)
{
   memset(node->stats, 0, sizeof node->stats);
   return 0;
}

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

static int dnxNodeListBuilder(DnxNodeData * node, void * data)
{
   // only return node information if node has been active since last reset
   return node->stats[REQUESTS_RECEIVED]? 
         appendString((char **)data, "%s,", node->addrstr): 0;
}

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

static char * buildMgmtNodeListReply(void)
{
   char * rsp = 0;   

   dnxStatsForEachNode(dnxNodeListBuilder, &rsp);

   // remove trailing comma in non-empty response
   if (rsp)
   {
      size_t len = strlen(rsp);
      if (len && rsp[len - 1] == ',') rsp[len - 1] = 0;
   }
   return rsp? rsp: xstrdup("");
}

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

static char * buildMgmtCfgReply(void)
{
   char * buf;
   size_t bufsz = 0;

   // first call gets size
   if (dnxCfgParserGetCfg(s_parser, 0, &bufsz) != 0)
      return 0;

   // second call gets configuration
   if ((buf = (char *)xmalloc(bufsz)) != 0
         && dnxCfgParserGetCfg(s_parser, buf, &bufsz) != 0)
      xfree(buf), (buf = 0);

   return buf;
}

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

static void * dnxAgentServer(void * data)
{
   int ret;
   DnxMgmtRequest Msg;
   Msg.action = 0;

   dnxLog("DNX Server Agent awaiting commands...");

   while (!s_shutdown)
   {
      memset(Msg.address, '\0', DNX_MAX_ADDRESS);

      // wait 2 second for a request; process the request, if valid
      if ((ret = dnxWaitForMgmtRequest(s_agent, &Msg, Msg.address, 2)) == DNX_OK)
      {
         DnxMgmtReply Rsp;
         char addrstr[DNX_MAX_ADDRSTR];

         dnxDebug(2, "Received MgmtRequest from %s.", 
               dnxNtop(Msg.address, addrstr, sizeof addrstr));

         // setup some default response values
         Rsp.xid = Msg.xid;
         Rsp.status = DNX_REQ_ACK;
         Rsp.reply = 0;

         // perform the requested action
         if (!strcmp(Msg.action, "RECONFIGURE"))
         {
            s_reconfig = 1;
            Rsp.reply = xstrdup("OK");
         }
         else if (!strcmp(Msg.action, "RESETSTATS"))
         {
            dnxStatsResetServerStats();
            dnxStatsForEachNode(dnxResetNodeStats, 0);
            Rsp.reply = xstrdup("OK");
         }
         else if (!strncmp(Msg.action, "GETSTATS ", 9))
         {
            if ((Rsp.reply = buildMgmtStatsReply(Msg.action + 9)) == 0)
               Rsp.status = DNX_REQ_NAK;
         }
         else if (!strncmp(Msg.action, "GETNODESTATS ", 13))
         {
            if ((Rsp.reply = buildMgmtNodeStatsReply(Msg.action + 13)) == 0)
               Rsp.status = DNX_REQ_NAK;
         }
         else if (!strcmp(Msg.action, "GETNODELIST"))
         {
            if ((Rsp.reply = buildMgmtNodeListReply()) == 0)
               Rsp.status = DNX_REQ_NAK;
         }
         else if (!strcmp(Msg.action, "GETCONFIG"))
         {
            if ((Rsp.reply = buildMgmtCfgReply()) == 0)
               Rsp.status = DNX_REQ_NAK;
         }
         else if (!strcmp(Msg.action, "GETVERSION"))
         {
            if ((Rsp.reply = versionText()) == 0)
               Rsp.status = DNX_REQ_NAK;
         }
         else if (!strcmp(Msg.action, "HELP"))
         {
            if ((Rsp.reply = buildHelpReply()) == 0)
               Rsp.status = DNX_REQ_NAK;
         }

         // send response, log response failures
         if ((ret = dnxSendMgmtReply(s_agent, &Rsp, Msg.address)) != 0)
            dnxLog("Agent response failure: %s.", dnxErrorString(ret));

         // free request and reply message buffers
         xfree(Rsp.reply);
         xfree(Msg.action);
      }
      else if (ret != DNX_ERR_TIMEOUT)
         dnxLog("Agent channel failure: %s.", dnxErrorString(ret));

      if (s_reconfig)
      {
         dnxReconfigure();
         s_reconfig = 0;
      }
   }

   dnxLog("Agent terminating...");

   return 0;
}

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

void dnxAgentSignalReconfig(void)
{
   s_reconfig = 1;
}

int dnxInitAgent(char * agentUrl, DnxCfgParser * parser)
{
   int ret;

   s_shutdown = 0;
   s_agentTid = 0;
   s_parser = parser;

   if ((ret = dnxChanMapAdd(s_agentName, agentUrl)) != DNX_OK)
      dnxLog("AGENT channel init failed: %s.", dnxErrorString(ret));
   else if ((ret = dnxConnect(s_agentName, DNX_MODE_PASSIVE, &s_agent)) != DNX_OK)
   {
      dnxLog("AGENT channel connect failed: %s.", dnxErrorString(ret));
      dnxChanMapDelete(s_agentName);
   }
   else if ((ret = pthread_create(&s_agentTid, 0, dnxAgentServer, 0)) != 0)
   {
      dnxLog("AGENT server init failed: %s.", strerror(ret));
      dnxDisconnect(s_agent);
      dnxChanMapDelete(s_agentName);
      ret = DNX_ERR_THREAD;
   }
   return ret;
}

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

void dnxReleaseAgent(void)
{
   if (s_agentTid)
   {
      s_shutdown = 1;
      pthread_join(s_agentTid, 0);
      dnxDisconnect(s_agent);
      dnxChanMapDelete(s_agentName);
   }
}

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