agent.c

Go to the documentation of this file.

/*
 SPDX-FileCopyrightText: © 2010, 2011, 2012 Hewlett-Packard Development Company, L.P.
 SPDX-FileCopyrightText: © 2015 Siemens AG
 
 SPDX-License-Identifier: GPL-2.0-only
*/
 
/* local includes */
#include <agent.h>
#include <database.h>
#include <event.h>
#include <host.h>
#include <job.h>
#include <logging.h>
#include <scheduler.h>
 
/* library includes */
#include <limits.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
 
/* unix library includes */
#include <fcntl.h>
#include <limits.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
 
/* other library includes */
#include <glib.h>
 
#define TEST_NULV(a) if(!a) { \
  errno = EINVAL; ERROR("agent passed is NULL, cannot proceed"); return; }
#define TEST_NULL(a, ret) if(!a) { \
  errno = EINVAL; ERROR("agent passed is NULL, cannot proceed"); return ret; }
 
#define AGENT_CREDENTIAL                                                 \
  log_printf("JOB[%d].%s[%d.%s]: ", agent->owner->id, agent->type->name, \
      agent->pid, agent->host->name)
 
#define AGENT_LOG_CREDENTIAL                                              \
  con_printf(job_log(agent->owner), "JOB[%d].%s[%d.%s]: ",                \
      agent->owner->id, agent->type->name, agent->pid, agent->host->name)
 
#define AGENT_ERROR(...) do {                       \
  log_printf("ERROR: %s.%d: ", __FILE__, __LINE__); \
  AGENT_CREDENTIAL;                                 \
  log_printf(__VA_ARGS__);                          \
  log_printf("\n"); } while(0)
 
#define AGENT_NOTIFY(...) if(TEST_NOTIFY) do { \
  log_printf("NOTE: ");                           \
  AGENT_CREDENTIAL;                            \
  log_printf(__VA_ARGS__);                        \
  log_printf("\n"); } while(0)
 
#define AGENT_WARNING(...) if(TEST_WARNING) do {      \
  log_printf("WARNING %s.%d: ", __FILE__, __LINE__);  \
  AGENT_CREDENTIAL;                                   \
  log_printf(__VA_ARGS__);                            \
  log_printf("\n"); } while(0)
 
#define AGENT_SEQUENTIAL_PRINT(...) if(TVERB_AGENT) do { \
  AGENT_CREDENTIAL;                          \
  log_printf(__VA_ARGS__); } while(0)
 
#define AGENT_CONCURRENT_PRINT(...) do {                             \
  AGENT_LOG_CREDENTIAL;                                 \
  con_printf(job_log(agent->owner), __VA_ARGS__); } while(0)
 
/* ************************************************************************** */
/* **** Data Types ********************************************************** */
/* ************************************************************************** */
 
#define SELECT_STRING(passed) MK_STRING_LIT(AGENT_##passed),
const char* agent_status_strings[] =
{ AGENT_STATUS_TYPES(SELECT_STRING) };
#undef SELECT_STRING
 
/* ************************************************************************** */
/* **** Local Functions ***************************************************** */
/* ************************************************************************** */
 
static int agent_close_fd(int* pid_ptr, agent_t* agent, agent_t* excepted)
{
  TEST_NULL(agent, 0);
  if (agent != excepted)
  {
    close(agent->from_child);
    close(agent->to_child);
    fclose(agent->read);
    fclose(agent->write);
  }
  return 0;
}
 
static int update(int* pid_ptr, agent_t* agent, gpointer unused)
{
  TEST_NULL(agent, 0);
  int nokill = is_agent_special(agent, SAG_NOKILL) || is_meta_special(agent->type, SAG_NOKILL);
 
  if (agent->status == AG_SPAWNED || agent->status == AG_RUNNING || agent->status == AG_PAUSED)
  {
    /* check last checkin time */
    if (time(NULL) - agent->check_in > CONF_agent_death_timer && !(agent->owner->status == JB_PAUSED) && !nokill)
    {
      AGENT_CONCURRENT_PRINT("no heartbeat for %d seconds\n", (time(NULL) - agent->check_in));
      agent_kill(agent);
      return 0;
    }
 
    /* check items processed */
    if (agent->status != AG_PAUSED && !agent->alive)
    {
      agent->n_updates++;
    }
    else
    {
      agent->n_updates = 0;
    }
    if (agent->n_updates > CONF_agent_update_number && !nokill)
    {
      AGENT_CONCURRENT_PRINT("agent has not set the alive flag in at least 10 minutes, killing\n");
      agent_kill(agent);
      return 0;
    }
 
    AGENT_SEQUENTIAL_PRINT("agent updated correctly, processed %d items: %d\n", agent->total_analyzed,
        agent->n_updates);
    agent->alive = 0;
  }
 
  return 0;
}
 
static int agent_kill_traverse(int* pid, agent_t* agent, gpointer unused)
{
  agent_kill(agent);
  return FALSE;
}
 
static int agent_list(char* name, meta_agent_t* ma, GOutputStream* ostr)
{
  if (ma->valid)
  {
    g_output_stream_write(ostr, name, strlen(name), NULL, NULL);
    g_output_stream_write(ostr, " ", 1, NULL, NULL);
  }
  return FALSE;
}
 
static int agent_test(const gchar* name, meta_agent_t* ma, scheduler_t* scheduler)
{
  static int32_t id_gen = -1;
 
  GList* iter;
  host_t* host;
  char *jq_cmd_args = 0;
 
  for (iter = scheduler->host_queue; iter != NULL; iter = iter->next)
  {
    host = (host_t*) iter->data;
    V_AGENT("META_AGENT[%s] testing on HOST[%s]\n", ma->name, host->name);
    job_t* job = job_init(scheduler->job_list, scheduler->job_queue, ma->name, host->name, id_gen--, 0, 0, 0, 0, jq_cmd_args);
    agent_init(scheduler, host, job);
  }
 
  return 0;
}
 
static void agent_listen(scheduler_t* scheduler, agent_t* agent)
{
  /* static locals */
#if GLIB_MAJOR_VERSION >= 2 && GLIB_MINOR_VERSION >= 32
  static GMutex version_lock;
#else
  static GStaticMutex version_lock = G_STATIC_MUTEX_INIT;
#endif
 
  /* locals */
  char buffer[1024]; // buffer to store c strings read from agent
  GMatchInfo* match; // regex match information
  char* arg;         // used during regex retrievals
  int relevant;      // used during special retrievals
 
  TEST_NULV(agent);
 
  if (fgets(buffer, sizeof(buffer), agent->read) == NULL)
  {
    AGENT_CONCURRENT_PRINT("pipe from child closed: %s\n", strerror(errno));
    g_thread_exit(NULL);
  }
 
  /* check to make sure "VERSION" was sent */
  buffer[strlen(buffer) - 1] = '\0';
  if (strncmp(buffer, "VERSION: ", 9) != 0)
  {
    if (strncmp(buffer, "@@@1", 4) == 0)
    {
      THREAD_FATAL(job_log(agent->owner), "agent crashed before sending version information");
    }
    else
    {
      agent->type->valid = 0;
      agent_fail_event(scheduler, agent);
      agent_kill(agent);
      con_printf(main_log, "ERROR %s.%d: agent %s.%s has been invalidated, removing from agents\n", __FILE__, __LINE__,
          agent->host->name, agent->type->name);
      AGENT_CONCURRENT_PRINT("agent didn't send version information: \"%s\"\n", buffer);
      return;
    }
  }
 
  /* check that the VERSION information is correct */
#if GLIB_MAJOR_VERSION >= 2 && GLIB_MINOR_VERSION >= 32
  g_mutex_lock(&version_lock);
#else
  g_static_mutex_lock(&version_lock);
#endif
  strcpy(buffer, &buffer[9]);
  if (agent->type->version == NULL && agent->type->valid)
  {
    agent->type->version_source = agent->host->name;
    agent->type->version = g_strdup(buffer);
    if (TVERB_AGENT)
      con_printf(main_log, "META_AGENT[%s.%s] version is: \"%s\"\n", agent->host->name, agent->type->name,
          agent->type->version);
  }
  else if (strcmp(agent->type->version, buffer) != 0)
  {
    con_printf(job_log(agent->owner), "ERROR %s.%d: META_DATA[%s] invalid agent spawn check\n", __FILE__, __LINE__,
        agent->type->name);
    con_printf(job_log(agent->owner), "ERROR: versions don't match: %s(%s) != received: %s(%s)\n",
        agent->type->version_source, agent->type->version, agent->host->name, buffer);
    agent->type->valid = 0;
    agent_kill(agent);
#if GLIB_MAJOR_VERSION >= 2 && GLIB_MINOR_VERSION >= 32
    g_mutex_unlock(&version_lock);
#else
    g_static_mutex_unlock(&version_lock);
#endif
    return;
  }
#if GLIB_MAJOR_VERSION >= 2 && GLIB_MINOR_VERSION >= 32
  g_mutex_unlock(&version_lock);
#else
  g_static_mutex_unlock(&version_lock);
#endif
 
  while (1)
  {
    /* get message from agent */
    if (fgets(buffer, sizeof(buffer), agent->read) == NULL)
      g_thread_exit(NULL);
 
    buffer[strlen(buffer) - 1] = '\0';
 
    if (strlen(buffer) == 0)
      continue;
 
    if (TVERB_AGENT && (TVERB_SPECIAL || strncmp(buffer, "SPECIAL", 7) != 0))
      AGENT_CONCURRENT_PRINT("received: \"%s\"\n", buffer);
 
    if (strncmp(buffer, "BYE", 3) == 0)
    {
      if ((agent->return_code = atoi(&(buffer[4]))) != 0)
      {
        AGENT_CONCURRENT_PRINT("agent failed with error code %d\n", agent->return_code);
        event_signal(agent_fail_event, agent);
      }
      break;
    }
 
    if (strncmp(buffer, "@@@1", 4) == 0)
      break;
 
    if (strncmp(buffer, "@@@0", 4) == 0 && agent->updated)
    {
      aprintf(agent, "%s\n", agent->data);
      aprintf(agent, "END\n");
      fflush(agent->write);
      agent->updated = 0;
      continue;
    }
 
    /* agent just checked in */
    agent->check_in = time(NULL);
 
    if (strncmp(buffer, "OK", 2) == 0)
    {
      if (agent->status != AG_PAUSED)
        event_signal(agent_ready_event, agent);
    }
 
    else if (strncmp(buffer, "HEART", 5) == 0)
    {
      g_regex_match(scheduler->parse_agent_msg, buffer, 0, &match);
 
      arg = g_match_info_fetch(match, 3);
      agent->total_analyzed = atoi(arg);
      g_free(arg);
 
      arg = g_match_info_fetch(match, 6);
      agent->alive = (arg[0] == '1' || agent->alive);
      g_free(arg);
 
      g_match_info_free(match);
      match = NULL;
 
      database_job_processed(agent->owner->id, agent->total_analyzed);
    }
 
    else if (strncmp(buffer, "EMAIL", 5) == 0)
    {
      agent->owner->message = g_strdup(buffer + 6);
    }
 
    else if (strncmp(buffer, "SPECIAL", 7) == 0)
    {
      relevant = INT_MAX;
 
      g_regex_match(scheduler->parse_agent_msg, buffer, 0, &match);
 
      arg = g_match_info_fetch(match, 3);
      relevant &= atoi(arg);
      g_free(arg);
 
      arg = g_match_info_fetch(match, 6);
      if (atoi(arg))
      {
        if (agent->special & relevant)
          relevant = 0;
      }
      else
      {
        if (!(agent->special & relevant))
          relevant = 0;
      }
      g_free(arg);
 
      g_match_info_free(match);
 
      agent->special ^= relevant;
    }
 
    else if (strncmp(buffer, "GETSPECIAL", 10) == 0)
    {
      g_regex_match(scheduler->parse_agent_msg, buffer, 0, &match);
 
      arg = g_match_info_fetch(match, 3);
      relevant = atoi(arg);
      g_free(arg);
 
      if (agent->special & relevant)
        aprintf(agent, "VALUE: 1\n");
      else
        aprintf(agent, "VALUE: 0\n");
 
      g_match_info_free(match);
    }
 
    else if (!(TVERB_AGENT))
      AGENT_CONCURRENT_PRINT("\"%s\"\n", buffer);
  }
 
  if (TVERB_AGENT)
    AGENT_CONCURRENT_PRINT("communication thread closing\n");
}
 
static void shell_parse(char* confdir, int user_id, int group_id, char* input, char *jq_cmd_args, int jobId, int* argc, char*** argv)
{
  char* begin;
  char* curr;
  int idx = 0;
#define MAX_CMD_ARGS 30
 
  *argv = g_new0(char*, MAX_CMD_ARGS);
  begin = NULL;
 
  for (curr = input; *curr; curr++)
  {
    if (*curr == ' ')
    {
      if (begin == NULL)
        continue;
 
      if (*begin == '"')
        continue;
 
      *curr = '\0';
      (*argv)[idx++] = g_strdup(begin);
      begin = NULL;
    }
    else if (begin == NULL)
    {
      begin = curr;
    }
    else if (*begin == '"' && *curr == '"')
    {
      *begin = '\0';
      *curr = '\0';
 
      (*argv)[idx++] = g_strdup(begin + 1);
      begin = NULL;
    }
    if (idx > MAX_CMD_ARGS - 7)
      break;
  }
 
  (*argv)[idx++] = g_strdup_printf("--jobId=%d", jobId);
  (*argv)[idx++] = g_strdup_printf("--config=%s", confdir);
  (*argv)[idx++] = g_strdup_printf("--userID=%d", user_id);
  (*argv)[idx++] = g_strdup_printf("--groupID=%d", group_id);
  (*argv)[idx++] = "--scheduler_start";
  if (jq_cmd_args)
  {
    const char *start = jq_cmd_args;
    const char *current = jq_cmd_args;
    gboolean in_quotes = FALSE;
 
    while (*current != '\0')
    {
      if (*current == '\'' || *current == '"')
        in_quotes = !in_quotes;
      else if (*current == ' ' && !in_quotes)
      {
        if (current > start)
        {
          int len = current - start;
          char *arg = g_strndup(start, len);
          (*argv)[idx++] = arg;
        }
        start = current + 1;
      }
      current++;
    }
 
    if (current > start)
    {
      char *arg = g_strndup(start, current - start);
      (*argv)[idx++] = arg;
    }
  }
  (*argc) = idx;
}
 
typedef struct
{
  scheduler_t* scheduler; 
  agent_t* agent;         
} agent_spawn_args;
 
static void* agent_spawn(agent_spawn_args* pass)
{
  /* locals */
  scheduler_t* scheduler = pass->scheduler;
  agent_t* agent = pass->agent;
  gchar* tmp;                 // pointer to temporary string
  gchar** args;               // the arguments that will be passed to the child
  int argc;                   // the number of arguments parsed
  int len;
  char buffer[2048];          // character buffer
 
  /* spawn the new process */
  while ((agent->pid = fork()) < 0)
    sleep(rand() % CONF_fork_backoff_time);
 
  /* we are in the child */
  if (agent->pid == 0)
  {
    /* set the child's stdin and stdout to use the pipes */
    dup2(agent->from_parent, fileno(stdin));
    dup2(agent->to_parent, fileno(stdout));
    dup2(agent->to_parent, fileno(stderr));
 
    /* close all the unnecessary file descriptors */
    g_tree_foreach(scheduler->agents, (GTraverseFunc) agent_close_fd, agent);
    close(agent->from_child);
    close(agent->to_child);
 
    /* set the priority of the process to the job's priority */
    if (nice(agent->owner->priority) == -1)
      ERROR("unable to correctly set priority of agent process %d", agent->pid);
 
    /* if host is null, the agent will run locally to */
    /* run the agent locally, use the commands that    */
    /* were parsed when the meta_agent was created    */
    if (strcmp(agent->host->address, LOCAL_HOST) == 0)
    {
      shell_parse(scheduler->sysconfigdir, agent->owner->user_id, agent->owner->group_id,
                  agent->type->raw_cmd, agent->owner->jq_cmd_args,
                  agent->owner->parent_id, &argc, &args);
 
      tmp = args[0];
      args[0] = g_strdup_printf(AGENT_BINARY, scheduler->sysconfigdir,
      AGENT_CONF, agent->type->name, tmp);
 
      strcpy(buffer, args[0]);
      *strrchr(buffer, '/') = '\0';
      if (chdir(buffer) != 0)
      {
        ERROR("unable to change working directory: %s\n", strerror(errno));
      }
 
      execv(args[0], args);
    }
    /* otherwise the agent will be started using ssh   */
    /* if the agent is started using ssh we don't need */
    /* to fully parse the arguments, just pass the run */
    /* command as the last argument to the ssh command */
    else
    {
      args = g_new0(char*, 5);
      len = snprintf(buffer, sizeof(buffer), AGENT_BINARY " --userID=%d --groupID=%d --scheduler_start --jobId=%d",
                     agent->host->agent_dir, AGENT_CONF, agent->type->name, agent->type->raw_cmd,
                     agent->owner->user_id, agent->owner->group_id, agent->owner->parent_id);
 
      if (len>=sizeof(buffer)) {
        *(buffer + sizeof(buffer) - 1) = '\0';
        log_printf("ERROR %s.%d: JOB[%d.%s]: exec failed: truncated buffer: \"%s\"",
            __FILE__, __LINE__, agent->owner->id, agent->owner->agent_type, buffer);
 
        exit(5);
      }
 
      args[0] = "/usr/bin/ssh";
      args[1] = agent->host->address;
      args[2] = buffer;
      args[3] = agent->owner->jq_cmd_args;
      args[4] = NULL;
      execv(args[0], args);
    }
 
    /* If we reach here, the exec call has failed */
    log_printf("ERROR %s.%d: JOB[%d.%s]: exec failed: pid = %d, errno = \"%s\"", __FILE__, __LINE__, agent->owner->id,
        agent->owner->agent_type, getpid(), strerror(errno));
  }
  /* we are in the parent */
  else
  {
    event_signal(agent_create_event, agent);
    agent_listen(scheduler, agent);
  }
 
  return NULL;
}
 
/* ************************************************************************** */
/* **** Constructor Destructor ********************************************** */
/* ************************************************************************** */
 
meta_agent_t* meta_agent_init(char* name, char* cmd, int max, int spc)
{
  /* locals */
  meta_agent_t* ma;
 
  /* test inputs */
  if (!name || !cmd)
  {
    ERROR("invalid arguments passed to meta_agent_init()");
    return NULL;
  }
 
  /* confirm valid inputs */
  if (strlen(name) > MAX_NAME || strlen(cmd) > MAX_CMD)
  {
    log_printf("ERROR failed to load %s meta agent", name);
    return NULL;
  }
 
  /* inputs are valid, create the meta_agent */
  ma = g_new0(meta_agent_t, 1);
 
  strcpy(ma->name, name);
  strcpy(ma->raw_cmd, cmd);
  strcat(ma->raw_cmd, " --scheduler_start");
  ma->max_run = max;
  ma->run_count = 0;
  ma->special = spc;
  ma->version = NULL;
  ma->valid = TRUE;
 
  return ma;
}
 
void meta_agent_destroy(meta_agent_t* ma)
{
  TEST_NULV(ma);
  g_free(ma->version);
  g_free(ma);
}
 
agent_t* agent_init(scheduler_t* scheduler, host_t* host, job_t* job)
{
  /* local variables */
  agent_t* agent;
  int child_to_parent[2];
  int parent_to_child[2];
  agent_spawn_args* pass;
 
  /* check job input */
  if (!job)
  {
    log_printf("ERROR %s.%d: NULL job passed to agent init\n", __FILE__, __LINE__);
    log_printf("ERROR: no other information available\n");
    return NULL;
  }
 
  /* check that the agent type exists */
  if (g_tree_lookup(scheduler->meta_agents, job->agent_type) == NULL)
  {
    log_printf("ERROR %s.%d: jq_pk %d jq_type %s does not match any module in mods-enabled\n", __FILE__, __LINE__,
        job->id, job->agent_type);
    job->message = NULL;
    job_fail_event(scheduler, job);
    job_remove_agent(job, scheduler->job_list, NULL);
    return NULL;
  }
 
  /* allocate memory and do trivial assignments */
  agent = g_new(agent_t, 1);
  agent->type = g_tree_lookup(scheduler->meta_agents, job->agent_type);
  agent->status = AG_CREATED;
 
  /* make sure that there is a metaagent for the job */
  if (agent->type == NULL)
  {
    ERROR("meta agent %s does not exist", job->agent_type);
    return NULL;
  }
 
  /* check if the agent is valid */
  if (!agent->type->valid)
  {
    ERROR("agent %s has been invalidated by version information", job->agent_type);
    return NULL;
  }
 
  /* create the pipes between the child and the parent */
  if (pipe(parent_to_child) != 0)
  {
    ERROR("JOB[%d.%s] failed to create parent to child pipe", job->id, job->agent_type);
    g_free(agent);
    return NULL;
  }
  if (pipe(child_to_parent) != 0)
  {
    ERROR("JOB[%d.%s] failed to create child to parent pipe", job->id, job->agent_type);
    g_free(agent);
    return NULL;
  }
 
  /* set file identifiers to correctly talk to children */
  agent->from_parent = parent_to_child[0];
  agent->to_child = parent_to_child[1];
  agent->from_child = child_to_parent[0];
  agent->to_parent = child_to_parent[1];
 
  /* initialize other info */
  agent->host = host;
  agent->owner = job;
  agent->updated = 0;
  agent->n_updates = 0;
  agent->data = NULL;
  agent->return_code = -1;
  agent->total_analyzed = 0;
  agent->special = 0;
 
  /* open the relevant file pointers */
  if ((agent->read = fdopen(agent->from_child, "r")) == NULL)
  {
    ERROR("JOB[%d.%s] failed to initialize read file", job->id, job->agent_type);
    g_free(agent);
    return NULL;
  }
  if ((agent->write = fdopen(agent->to_child, "w")) == NULL)
  {
    ERROR("JOB[%d.%s] failed to initialize write file", job->id, job->agent_type);
    g_free(agent);
    return NULL;
  }
 
  /* increase the load on the host and count of running agents */
  if (agent->owner->id > 0)
  {
    host_increase_load(agent->host);
    meta_agent_increase_count(agent->type);
  }
 
  /* spawn the listen thread */
  pass = g_new0(agent_spawn_args, 1);
  pass->scheduler = scheduler;
  pass->agent = agent;
 
#if GLIB_MAJOR_VERSION >= 2 && GLIB_MINOR_VERSION >= 32
  agent->thread = g_thread_new(agent->type->name, (GThreadFunc) agent_spawn, pass);
#else
  agent->thread = g_thread_create((GThreadFunc)agent_spawn, pass, 1, NULL);
#endif
 
  return agent;
}
 
void agent_destroy(agent_t* agent)
{
  TEST_NULV(agent);
 
  /* close all of the files still open for this agent */
  close(agent->from_child);
  close(agent->to_child);
  close(agent->from_parent);
  close(agent->to_parent);
  fclose(agent->write);
  fclose(agent->read);
 
  /* release the child process */
  g_free(agent);
}
 
/* ************************************************************************** */
/* **** Events ************************************************************** */
/* ************************************************************************** */
 
void agent_death_event(scheduler_t* scheduler, pid_t* pid)
{
  agent_t* agent;
  int status = pid[1];
 
  if ((agent = g_tree_lookup(scheduler->agents, &pid[0])) == NULL)
  {
    ERROR("invalid agent death event: pid[%d]", pid[0]);
    return;
  }
 
  if (agent->owner->id >= 0)
    event_signal(database_update_event, NULL);
 
  if (write(agent->to_parent, "@@@1\n", 5) != 5)
    AGENT_SEQUENTIAL_PRINT("write to agent unsuccessful: %s\n", strerror(errno));
  g_thread_join(agent->thread);
 
  if (agent->return_code != 0)
  {
    if (WIFEXITED(status))
    {
      AGENT_CONCURRENT_PRINT("agent failed, code: %d\n", (status >> 8));
    }
    else if (WIFSIGNALED(status))
    {
      AGENT_CONCURRENT_PRINT("agent was killed by signal: %d.%s\n", WTERMSIG(status), strsignal(WTERMSIG(status)));
      if (WCOREDUMP(status))
        AGENT_CONCURRENT_PRINT("agent produced core dump\n");
    }
    else
    {
      AGENT_CONCURRENT_PRINT("agent failed, code: %d\n", agent->return_code);
    }
    AGENT_WARNING("agent closed unexpectedly, agent status was %s", agent_status_strings[agent->status]);
    agent_fail_event(scheduler, agent);
  }
 
  if (agent->status != AG_PAUSED && agent->status != AG_FAILED)
    agent_transition(agent, AG_PAUSED);
 
  job_update(scheduler, agent->owner);
  if (agent->status == AG_FAILED && agent->owner->id < 0)
  {
    log_printf("ERROR %s.%d: agent %s.%s has failed scheduler startup test\n", __FILE__, __LINE__, agent->host->name,
        agent->type->name);
    agent->type->valid = 0;
  }
 
  if (agent->owner->id < 0 && !agent->type->valid)
    AGENT_SEQUENTIAL_PRINT("agent failed startup test, removing from meta agents\n");
 
  AGENT_SEQUENTIAL_PRINT("successfully remove from the system\n");
  job_remove_agent(agent->owner, scheduler->job_list, agent);
  g_tree_remove(scheduler->agents, &agent->pid);
  g_free(pid);
}
 
void agent_create_event(scheduler_t* scheduler, agent_t* agent)
{
  TEST_NULV(agent);
 
  AGENT_SEQUENTIAL_PRINT("agent successfully spawned\n");
  g_tree_insert(scheduler->agents, &agent->pid, agent);
  agent_transition(agent, AG_SPAWNED);
  job_add_agent(agent->owner, agent);
}
 
void agent_ready_event(scheduler_t* scheduler, agent_t* agent)
{
  int ret;
 
  TEST_NULV(agent);
  if (agent->status == AG_SPAWNED)
  {
    agent_transition(agent, AG_RUNNING);
    AGENT_SEQUENTIAL_PRINT("agent successfully created\n");
  }
 
  if ((ret = job_is_open(scheduler, agent->owner)) == 0)
  {
    agent_transition(agent, AG_PAUSED);
    job_finish_agent(agent->owner, agent);
    job_update(scheduler, agent->owner);
    return;
  }
  else if (ret < 0)
  {
    agent_transition(agent, AG_FAILED);
    return;
  }
  else
  {
    agent->data = job_next(agent->owner);
    agent->updated = 1;
  }
 
  if (write(agent->to_parent, "@@@0\n", 5) != 5)
  {
    AGENT_ERROR("failed sending new data to agent");
    agent_kill(agent);
  }
}
 
void agent_update_event(scheduler_t* scheduler, void* unused)
{
  g_tree_foreach(scheduler->agents, (GTraverseFunc) update, NULL);
}
 
void agent_fail_event(scheduler_t* scheduler, agent_t* agent)
{
  TEST_NULV(agent);
  agent_transition(agent, AG_FAILED);
  job_fail_agent(agent->owner, agent);
  if (write(agent->to_parent, "@@@1\n", 5) != 5)
    AGENT_ERROR("Failed to kill agent thread cleanly");
}
 
void list_agents_event(scheduler_t* scheduler, GOutputStream* ostr)
{
  g_tree_foreach(scheduler->meta_agents, (GTraverseFunc) agent_list, ostr);
  g_output_stream_write(ostr, "\nend\n", 5, NULL, NULL);
}
 
/* ************************************************************************** */
/* **** Modifier Functions ************************************************** */
/* ************************************************************************** */
 
void agent_transition(agent_t* agent, agent_status new_status)
{
  AGENT_SEQUENTIAL_PRINT("agent status change: %s -> %s\n", agent_status_strings[agent->status],
      agent_status_strings[new_status]);
 
  if (agent->owner->id > 0)
  {
    if (agent->status == AG_PAUSED)
    {
      host_increase_load(agent->host);
      meta_agent_increase_count(agent->type);
    }
    if (new_status == AG_PAUSED)
    {
      host_decrease_load(agent->host);
      meta_agent_decrease_count(agent->type);
    }
  }
 
  agent->status = new_status;
}
 
void agent_pause(agent_t* agent)
{
  kill(agent->pid, SIGSTOP);
  agent_transition(agent, AG_PAUSED);
}
 
void agent_unpause(agent_t* agent)
{
  kill(agent->pid, SIGCONT);
  agent_transition(agent, AG_RUNNING);
}
 
void agent_print_status(agent_t* agent, GOutputStream* ostr)
{
  gchar* status_str;
  char time_buf[64];
  struct tm* time_info;
 
  TEST_NULV(agent);
  TEST_NULV(ostr);
 
  strcpy(time_buf, "(none)");
  time_info = localtime(&agent->check_in);
  if (time_info)
    strftime(time_buf, sizeof(time_buf), "%F %T", localtime(&agent->check_in));
  status_str = g_strdup_printf("agent:%d host:%s type:%s status:%s time:%s\n", agent->pid, agent->host->name,
      agent->type->name, agent_status_strings[agent->status], time_buf);
 
  AGENT_SEQUENTIAL_PRINT("AGENT_STATUS: %s", status_str);
  g_output_stream_write(ostr, status_str, strlen(status_str), NULL, NULL);
  g_free(status_str);
  return;
}
 
void agent_kill(agent_t* agent)
{
  AGENT_SEQUENTIAL_PRINT("KILL: sending SIGKILL to pid %d\n", agent->pid);
  meta_agent_decrease_count(agent->type);
  kill(agent->pid, SIGKILL);
}
 
int aprintf(agent_t* agent, const char* fmt, ...)
{
  va_list args;
  int rc;
  char* tmp;
 
  va_start(args, fmt);
  if (TVERB_AGENT)
  {
    tmp = g_strdup_vprintf(fmt, args);
    tmp[strlen(tmp) - 1] = '\0';
    AGENT_CONCURRENT_PRINT("sent to agent \"%s\"\n", tmp);
    rc = fprintf(agent->write, "%s\n", tmp);
    g_free(tmp);
  }
  else
  {
    rc = vfprintf(agent->write, fmt, args);
  }
  va_end(args);
  fflush(agent->write);
 
  return rc;
}
 
ssize_t agent_write(agent_t* agent, const void* buf, int count)
{
  return write(agent->to_parent, buf, count);
}
 
/* ************************************************************************** */
/* **** static functions and meta agents ************************************ */
/* ************************************************************************** */
 
void test_agents(scheduler_t* scheduler)
{
  g_tree_foreach(scheduler->meta_agents, (GTraverseFunc) agent_test, scheduler);
}
 
void kill_agents(scheduler_t* scheduler)
{
  g_tree_foreach(scheduler->agents, (GTraverseFunc) agent_kill_traverse, NULL);
}
 
int add_meta_agent(GTree* meta_agents, char* name, char* cmd, int max, int spc)
{
  meta_agent_t* ma;
 
  if (name == NULL)
    return 0;
 
  if (g_tree_lookup(meta_agents, name) == NULL)
  {
    if ((ma = meta_agent_init(name, cmd, max, spc)) == NULL)
      return 0;
    g_tree_insert(meta_agents, ma->name, ma);
    return 1;
  }
 
  return 0;
}
 
int is_meta_special(meta_agent_t* ma, int special_type)
{
  return (ma != NULL) && ((ma->special & special_type) != 0);
}
 
int is_agent_special(agent_t* agent, int special_type)
{
  return (agent != NULL) && ((agent->special & special_type) != 0);
}
 
void meta_agent_increase_count(meta_agent_t* ma)
{
  ma->run_count++;
  V_AGENT("AGENT[%s] run increased to %d\n", ma->name, ma->run_count);
}
 
void meta_agent_decrease_count(meta_agent_t* ma)
{
  ma->run_count--;
  V_AGENT("AGENT[%s] run decreased to %d\n", ma->name, ma->run_count);
}