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 do {                               \
  if(agent && agent->owner)                                 \
    log_printf("JOB[%d].%s[%d.%s]: ", agent->owner->id,     \
        agent->type ? agent->type->name : "?",              \
        agent->pid, agent->host ? agent->host->name : "?"); \
  else                                                      \
    log_printf("AGENT[%s][pid=%d][host=%s]: ",              \
        agent && agent->type ? agent->type->name : "?",     \
        agent ? agent->pid : -1,                            \
        agent && agent->host ? agent->host->name : "?");    \
} while(0)
 
#define AGENT_LOG_CREDENTIAL do {                                \
  if(agent && agent->owner)                                      \
    con_printf(job_log(agent->owner), "JOB[%d].%s[%d.%s]: ",     \
        agent->owner->id, agent->type ? agent->type->name : "?", \
        agent->pid, agent->host ? agent->host->name : "?");      \
} while(0)
 
#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 gboolean agent_close_fd(int* pid_ptr, agent_t* agent, agent_t* excepted)
{
  TEST_NULL(agent, FALSE);
  if (agent != excepted)
  {
    // Use raw close(): fclose() would flush stdio buffers into the parent's pipes.
    close(agent->from_child);
    close(agent->to_child);
    close(agent->from_parent);
    close(agent->to_parent);
  }
  return FALSE;
}
 
static int update(int* pid_ptr, agent_t* agent, gpointer unused)
{
  TEST_NULL(agent, 0);
  if (agent->owner == NULL)
  {
    log_printf("ERROR %s.%d: Agent pid %d has no owner; killing to prevent NULL deref\n", __FILE__, __LINE__, agent->pid);
    agent_kill(agent);
    return 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;
}
 
/* Protects meta_agent_t::version and version_source - held by both the
 * agent_listen spawn thread and the event-loop thread (agent_meta_version_reset). */
#if GLIB_CHECK_VERSION(2, 32, 0)
static GMutex version_lock;
#else
static GStaticMutex version_lock = G_STATIC_MUTEX_INIT;
#endif
 
void agent_meta_version_reset(meta_agent_t* ma)
{
  if (ma == NULL) return;
#if GLIB_CHECK_VERSION(2, 32, 0)
  g_mutex_lock(&version_lock);
#else
  g_static_mutex_lock(&version_lock);
#endif
  /* Log the old version while holding the lock so the read and free are atomic. */
  if (ma->version != NULL)
    log_printf("NOTE: version refresh: resetting cached version for agent "
        "type \"%s\" (was \"%s\")\n", ma->name, ma->version);
  g_free(ma->version);
  ma->version = NULL;
  ma->version_source = NULL;
  ma->valid = TRUE;
#if GLIB_CHECK_VERSION(2, 32, 0)
  g_mutex_unlock(&version_lock);
#else
  g_static_mutex_unlock(&version_lock);
#endif
}
 
typedef struct
{
  pid_t    pid;    
  agent_t* agent;  
} fail_pid_arg;
 
static void agent_listen(scheduler_t* scheduler, agent_t* agent)
{
  /* 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_CHECK_VERSION(2, 32, 0)
  g_mutex_lock(&version_lock);
#else
  g_static_mutex_lock(&version_lock);
#endif
  /* skip the "VERSION: " prefix */
  const char* version_str = buffer + 9;
  if (agent->type->version == NULL && agent->type->valid)
  {
    agent->type->version_source = agent->host->name;
    agent->type->version = g_strdup(version_str);
    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, version_str) != 0)
  {
    /* Version mismatch: hard-fail during startup tests (id<0); adopt new version at runtime. */
    if (agent->owner != NULL && agent->owner->id < 0)
    {
      /* Startup test: hard-fail so version skew is caught at boot. */
      con_printf(job_log(agent->owner),
          "ERROR %s.%d: META_DATA[%s] invalid agent spawn check (startup)\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, version_str);
      agent->type->valid = 0;
      /* non-zero return_code so the startup-test job is marked failed */
      agent->return_code = -1;
      kill(-agent->pid, SIGKILL);
#if GLIB_CHECK_VERSION(2, 32, 0)
      g_mutex_unlock(&version_lock);
#else
      g_static_mutex_unlock(&version_lock);
#endif
      return;
    }
 
    /* Runtime: adopt new version and respawn cleanly. */
    con_printf(main_log,
        "WARNING %s.%d: META_AGENT[%s] version changed: \"%s\" -> \"%s\" "
        "(was reported by: %s, now: %s). Adopting new version; agent will respawn.\n",
        __FILE__, __LINE__, agent->type->name,
        agent->type->version, version_str,
        agent->type->version_source, agent->host->name);
 
    g_free(agent->type->version);
    agent->type->version = g_strdup(version_str);
    agent->type->version_source = agent->host->name;
    /* return_code=0 so the death event respawns cleanly instead of failing the job.
     * Kill the process group so child processes die too. */
    agent->return_code = 0;
    kill(-agent->pid, SIGKILL);
#if GLIB_CHECK_VERSION(2, 32, 0)
    g_mutex_unlock(&version_lock);
#else
    g_static_mutex_unlock(&version_lock);
#endif
    return;
  }
#if GLIB_CHECK_VERSION(2, 32, 0)
  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);
        /* Pass the pid, not the agent pointer: agent_death_event() can join this
         * thread and free the agent before the queued event runs. */
        fail_pid_arg* fail_arg = g_new0(fail_pid_arg, 1);
        fail_arg->pid = agent->pid;
        fail_arg->agent = agent;
        event_signal(agent_fail_event_pid, fail_arg);
      }
      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)
    {
      g_free(agent->owner->message);
      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;
  g_free(pass);  /* all values extracted; free before fork so g_thread_exit can't leak it */
  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 */
  if (agent->owner == NULL)
  {
    log_printf("ERROR %s.%d: Agent spawn requested but agent has no owner; aborting spawn.\n", __FILE__, __LINE__);
 
    /* Close FILE* streams if they were opened by agent_init(). */
    if (agent->read)  { fclose(agent->read);  agent->read = NULL; agent->from_child = -1; }
    if (agent->write) { fclose(agent->write); agent->write = NULL; agent->to_child   = -1; }
 
    /* from_child/to_child were already closed via fclose() above. */
    if (agent->from_parent >= 0) { close(agent->from_parent); agent->from_parent = -1; }
    if (agent->to_parent >= 0)   { close(agent->to_parent);   agent->to_parent   = -1; }
 
    agent->status = AG_FAILED;
    return NULL;
  }
 
  while ((agent->pid = fork()) < 0)
    sleep(rand() % CONF_fork_backoff_time);
 
  /* we are in the child */
  if (agent->pid == 0)
  {
    /* Own process group so kill(-pid) reaches all child processes. */
    setpgid(0, 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
  {
    /* Mirror child's setpgid to close the race between fork and exec. */
    setpgid(agent->pid, agent->pid);
 
    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 */
  meta_agent_t* ma = g_tree_lookup(scheduler->meta_agents, job->agent_type);
  if (ma == 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 = ma;
  agent->status = AG_CREATED;
 
  /* check if the agent is valid */
  if (!agent->type->valid)
  {
    ERROR("agent %s has been invalidated by version information", job->agent_type);
    /* Job will never get an agent; fail and remove it to avoid a leak. */
    g_free(agent);
    g_free(job->message);
    job->message = g_strdup("agent type has been invalidated");
    job_fail_event(scheduler, job);
    job_remove_agent(job, scheduler->job_list, NULL);
    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);
    close(parent_to_child[0]);
    close(parent_to_child[1]);
    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;
  agent->accounted = FALSE;
 
  /* 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);
    close(agent->from_child);
    close(agent->to_child);
    close(agent->from_parent);
    close(agent->to_parent);
    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);
    fclose(agent->read); /* closes from_child */
    close(agent->from_parent);
    close(agent->to_parent);
    close(agent->to_child);
    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);
    /* mark that this agent holds a slot, so it can be released even if the job
     * is removed before the death event runs */
    agent->accounted = TRUE;
  }
 
  /* spawn the listen thread */
  pass = g_new0(agent_spawn_args, 1);
  pass->scheduler = scheduler;
  pass->agent = agent;
 
#if GLIB_CHECK_VERSION(2, 32, 0)
  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);
 
  // from_child/to_child are owned by the FILE* wrappers; fclose closes them.
  close(agent->from_parent);
  close(agent->to_parent);
  if (agent->write)
  {
    fclose(agent->write);
  }
  if (agent->read)
  {
    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)
  {
    // SIGCHLD can race ahead of agent_create_event; retry up to 3 times.
    if (pid[2]++ < 3)
    {
      event_signal(agent_death_event, pid);
      return;
    }
    ERROR("invalid agent death event: pid[%d]", pid[0]);
    g_free(pid);
    return;
  }
 
  /* Ownerless agent: job_remove_agent already removed and freed the job, so the
   * usual decrement paths never ran. Unblock and join the thread, release the slot
   * this agent still holds, then drop it. */
  if (agent->owner == NULL)
  {
    log_printf("ERROR %s.%d: agent_death_event for ownerless agent pid %d - cleaning up\n",
        __FILE__, __LINE__, (int)pid[0]);
    if (write(agent->to_parent, "@@@1\n", 5) != 5)
    {
      AGENT_SEQUENTIAL_PRINT("write to ownerless agent unsuccessful: %s\n", strerror(errno));
    }
    g_thread_join(agent->thread);
    /* Release the slot if this agent was counted and still holds it (AG_PAUSED
     * agents already released theirs when they paused). */
    if (agent->accounted && agent->status != AG_PAUSED)
    {
      host_decrease_load(agent->host);
      meta_agent_decrease_count(agent->type);
      agent->accounted = FALSE;
    }
    g_tree_remove(scheduler->agents, &agent->pid);
    g_free(pid);
    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);
 
  /* Skip if the agent was already failed (the "BYE n" path may have run first);
   * calling it twice would add the agent to failed_agents twice. */
  if (agent->return_code != 0 && agent->status != AG_FAILED)
  {
    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);
  }
 
  /* Decrement run_count and host load once per death:
   *   AG_FAILED: agent_fail_event skipped it, so do it here.
   *   AG_PAUSED: already done at the PAUSED transition.
   *   otherwise: do it through agent_transition(AG_PAUSED). */
  if (agent->status == AG_FAILED)
  {
    if (agent->owner != NULL && agent->owner->id > 0)
    {
      host_decrease_load(agent->host);
      meta_agent_decrease_count(agent->type);
    }
  }
  else if (agent->status != AG_PAUSED)
  {
    agent_transition(agent, AG_PAUSED);
  }
 
  /* Move the dying agent to finished_agents so job_remove_agent can clean up.
   * If the job is still JB_CHECKEDOUT the agent died before it got any data
   * (e.g. version-mismatch respawn), so re-queue the job for a fresh dispatch
   * instead of completing it. */
  if (agent->owner != NULL &&
      g_list_find(agent->owner->running_agents, agent) != NULL)
  {
    if (agent->owner->status == JB_CHECKEDOUT)
    {
      agent->owner->running_agents = g_list_remove(
          agent->owner->running_agents, agent);
      /* reset the timer so the stale reaper gives it a fresh grace period */
      agent->owner->checkedout_at = time(NULL);
      g_sequence_insert_sorted(scheduler->job_queue, agent->owner,
          job_compare, NULL);
    }
    else
    {
      job_finish_agent(agent->owner, agent);
      job_update(scheduler, agent->owner);
    }
  }
  else if (agent->owner != NULL)
  {
    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 the agent has no job (owner is NULL), it shouldn't be here.
  // This prevents the "job passed is NULL
  if (agent->owner == NULL)
  {
      ERROR("Agent ready event received but agent has no owner. Terminating agent to prevent scheduler crash.");
      agent_kill(agent);
      return;
  }
 
  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)
  {
    /* DB lookup failed (connection lost or OOM). Fail the job and kill the agent:
     * the watchdog ignores AG_FAILED agents, so we must SIGKILL it ourselves or it
     * would run forever. agent_fail_event has already set AG_FAILED, so the later
     * agent_death_event will not fail it again. */
    agent_fail_event(scheduler, agent);
    job_update(scheduler, agent->owner);
    agent_kill(agent);
    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);
  }
}
 
typedef struct {
  GList* list; 
  time_t now;  
} zombie_ctx;
 
static gboolean collect_zombie_agents(int* pid_ptr, agent_t* agent, zombie_ctx* ctx)
{
  if (agent == NULL || agent->owner == NULL)
  {
    return FALSE;
  }
  // Job must be FAILED with agent still in running/spawned state.
  if (agent->owner->status != JB_FAILED)
  {
    return FALSE;
  }
  if (agent->status != AG_SPAWNED && agent->status != AG_RUNNING)
  {
    return FALSE;
  }
  // 3× timer guarantees the normal watchdog already attempted SIGKILL.
  if (ctx->now - agent->check_in <= (time_t)(3 * CONF_agent_death_timer))
  {
    return FALSE;
  }
 
  ctx->list = g_list_prepend(ctx->list, agent);
  return FALSE;
}
 
void agent_update_event(scheduler_t* scheduler, void* unused)
{
  g_tree_foreach(scheduler->agents, (GTraverseFunc) update, NULL);
 
  // Second pass: force-reap D-state zombie agents stuck in FAILED jobs.
  zombie_ctx zctx = { NULL, time(NULL) };
  g_tree_foreach(scheduler->agents, (GTraverseFunc) collect_zombie_agents, &zctx);
 
  GList* iter;
  for (iter = zctx.list; iter != NULL; iter = iter->next)
  {
    agent_t* zombie = (agent_t*)iter->data;
 
    log_printf("WARNING %s.%d: JOB[%d].%s[%d.%s]: agent unresponsive for %ld s"
               " in a FAILED job - forcing death-event cleanup\n",
               __FILE__, __LINE__,
               zombie->owner ? zombie->owner->id  : -1,
               zombie->type  ? zombie->type->name : "unknown",
               (int)zombie->pid,
               zombie->host  ? zombie->host->name : "unknown",
               (long)(zctx.now - zombie->check_in));
 
    /* Queue it rather than calling directly, which would block on g_thread_join.
     * This is a synthetic death (no real waitpid status), so pass status 0;
     * agent_death_event uses return_code, not pids[1], to decide the fail path. */
    pid_t* pids = g_new0(pid_t, 3);  /* [0]=pid [1]=status [2]=retry */
    pids[0] = zombie->pid;
    pids[1] = 0;
    event_signal(agent_death_event, pids);
  }
 
  g_list_free(zctx.list);
}
 
void agent_fail_event(scheduler_t* scheduler, agent_t* agent)
{
  TEST_NULV(agent);
 
  // Ownerless agent: job already removed; transition and unblock thread to avoid NULL deref.
  if (agent->owner == NULL)
  {
    AGENT_ERROR("agent_fail_event called on ownerless agent (pid=%d), killing cleanly", agent->pid);
    agent_transition(agent, AG_FAILED);
    if (write(agent->to_parent, "@@@1\n", 5) != 5)
    {
      AGENT_ERROR("Failed to kill ownerless agent thread cleanly");
    }
    return;
  }
 
  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 agent_fail_event_pid(scheduler_t* scheduler, void* arg)
{
  fail_pid_arg* fa = (fail_pid_arg*) arg;
  agent_t* agent = g_tree_lookup(scheduler->agents, &fa->pid);
 
  if (agent != NULL && agent == fa->agent && agent->status != AG_FAILED)
  {
    agent_fail_event(scheduler, agent);
  }
 
  g_free(fa);
}
 
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 != NULL && 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);
  /* kill the process group so child processes (e.g. sh/python3) die too.
   * Leave return_code alone: a non-zero value makes the job fail, not complete. */
  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);
}