ApMon.h

Go to the documentation of this file.

#ifdef _WIN32
  // FIXME: (MCl) the following warning tells that the usage of throw
  // declaration may cause trouble with VC > 7.1

  // Disable warning C4290: C++ exception specification ignored except to indicate a function is not __declspec(nothrow)
  #pragma warning ( disable : 4290 )
#endif

/*
 * ApMon - Application Monitoring Tool
 * Version: 2.2.0
 *
 * Copyright (C) 2006 California Institute of Technology
 *
 * Permission is hereby granted, free of charge, to use, copy and modify 
 * this software and its documentation (the "Software") for any
 * purpose, provided that existing copyright notices are retained in 
 * all copies and that this notice is included verbatim in any distributions
 * or substantial portions of the Software. 
 * This software is a part of the MonALISA framework (http://monalisa.cacr.caltech.edu).
 * Users of the Software are asked to feed back problems, benefits,
 * and/or suggestions about the software to the MonALISA Development Team
 * (developers@monalisa.cern.ch). Support for this software - fixing of bugs,
 * incorporation of new features - is done on a best effort basis. All bug
 * fixes and enhancements will be made available under the same terms and
 * conditions as the original software,

 * IN NO EVENT SHALL THE AUTHORS OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT
 * OF THE USE OF THIS SOFTWARE, ITS DOCUMENTATION, OR ANY DERIVATIVES THEREOF,
 * EVEN IF THE AUTHORS HAVE BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 * THE AUTHORS AND DISTRIBUTORS SPECIFICALLY DISCLAIM ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT. THIS SOFTWARE IS
 * PROVIDED ON AN "AS IS" BASIS, AND THE AUTHORS AND DISTRIBUTORS HAVE NO
 * OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR
 * MODIFICATIONS.
 */

#ifndef ApMon_h
#define ApMon_h

#include <sys/types.h>
#include <sys/stat.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdexcept>
#include <ctype.h>
#include <time.h>

#ifdef WIN32
#define NOMSG
#define NOGDI
#include <Winsock2.h>
#include <string.h>
#include <process.h>
#include "xdr.h"
#undef NOMSG
#undef NOGDI
#else
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <sys/utsname.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <net/if.h>
#include <rpc/rpc.h>
#include <netdb.h>
#include <unistd.h>
#include <pthread.h>
#include <pwd.h>
#include <grp.h>

#ifdef __APPLE__
#include <sys/param.h>
#else
#include <linux/param.h>
#endif

#endif  // ~WIN32

using namespace std;

#define XDR_STRING  0  
//#define XDR_INT16   1   // NOT SUPPORTED YET!
#define XDR_INT32   2  
//#define XDR_INT64   3    // THE SAME!
#define XDR_REAL32  4  
#define XDR_REAL64  5  
#define MAX_DGRAM_SIZE   8192  
#define MAX_STRING_LEN 512   
#define MAX_STRING_LEN1 (MAX_STRING_LEN + 1)
#define RET_SUCCESS  0  
#define RET_ERROR   -1  
#define PROCUTILS_ERROR -2
#define RET_NOT_SENT -3 
#define MAX_N_DESTINATIONS 30  
#define DEFAULT_PORT 8884 
#define MAX_HEADER_LENGTH 40  
#define FILE_INIT  1

#ifdef __APPLE__
#define OLIST_INIT  2
#else
#define LIST_INIT  2
#endif

#define DIRECT_INIT  3

#define JOB_MONITOR_INTERVAL 20

#define SYS_MONITOR_INTERVAL 20

#define RECHECK_INTERVAL 600

#define GEN_MONITOR_INTERVALS 10

#define MAX_MONITORED_JOBS 30

#define MAX_SYS_PARAMS 30

#define MAX_GEN_PARAMS 30

#define MAX_JOB_PARAMS 30

#define MAX_MSG_RATE 20

#define NLETTERS 26

#define TWO_BILLION 2000000000

#define APMON_VERSION "2.2.1"

typedef struct ConfURLs {
  int nConfURLs; 
  char *vURLs[MAX_N_DESTINATIONS];
  char *lastModifURLs[MAX_N_DESTINATIONS];
} ConfURLs;

typedef struct MonitoredJob {
  long pid;
  /* the job's working dierctory */
  char workdir[MAX_STRING_LEN];
  /* the cluster name that will be included in the monitoring datagrams */
  char clusterName[50]; 
  /* the node name that will be included in the monitoring datagrams */
  char nodeName[50];
} MonitoredJob;

#ifdef WIN32
#define pthread_mutex_lock(mutex_ref) (WaitForSingleObject(*mutex_ref, INFINITE))
#define pthread_mutex_unlock(mutex_ref) (ReleaseMutex(*mutex_ref))
#define pthread_mutex_destroy(mutex_ref) (CloseHandle(*mutex_ref))
#define pthread_cond_signal(event_ref) (SetEvent(*event_ref))
#define pthread_cond_destroy(event_ref) (CloseHandle(*event_ref))
#define ETIMEDOUT WAIT_TIMEOUT
#endif

class ApMon {
 protected:
  char *clusterName; 
  char *nodeName; 
  char *sysMonCluster; 
  char *sysMonNode;

  int nDestinations; 
  char **destAddresses; 
  int *destPorts; 
  char **destPasswds; 
  char *buf; 
  int dgramSize; 
#ifndef WIN32
  int sockfd; 
#else
  SOCKET sockfd;
#endif
 
  bool confCheck;

  int nInitSources;
  char  **initSources;
  /* The initialization type (from file / list / directly). */
  int initType;

  long recheckInterval;

  long crtRecheckInterval;

#ifndef WIN32

  pthread_t bkThread;
  
  pthread_mutex_t mutex;

  pthread_mutex_t mutexBack;
  
  pthread_mutex_t mutexCond;

  pthread_cond_t confChangedCond;
#else
 public:
  HANDLE bkThread;
  HANDLE mutex;
  HANDLE mutexBack;
  HANDLE mutexCond;
  HANDLE confChangedCond;
 protected:
#endif

  bool recheckChanged, jobMonChanged,sysMonChanged;

  bool haveBkThread;

  bool bkThreadStarted;

  bool stopBkThread;

  bool autoDisableMonitoring;

  bool sysMonitoring;

  bool jobMonitoring;

  bool genMonitoring;

  long jobMonitorInterval, sysMonitorInterval; 
  
  int genMonitorIntervals;

  int nSysMonitorParams, nJobMonitorParams, nGenMonitorParams;
  
  /* The names of the parameters that can be enabled/disabled by the user in
   * the system/job/general monitoring datagrams.
   */
  char *sysMonitorParams[MAX_SYS_PARAMS];
  char *genMonitorParams[MAX_GEN_PARAMS];
  char *jobMonitorParams[MAX_JOB_PARAMS];

  /* Arrays of flags that specifiy the active monitoring parameters (the
   * ones that are sent in the datagams). 
   */
  int actSysMonitorParams[MAX_SYS_PARAMS];
  int actGenMonitorParams[MAX_GEN_PARAMS];
  int actJobMonitorParams[MAX_JOB_PARAMS];

  ConfURLs confURLs;

  int nMonJobs;

  MonitoredJob *monJobs;

  long lastModifFile;

  /* The moment when the last datagram with job monitoring information
   * was sent. 
   */
  time_t lastJobInfoSend;

  char username[MAX_STRING_LEN];
  char groupname[MAX_STRING_LEN];
  char myHostname[MAX_STRING_LEN];
  char myIP[MAX_STRING_LEN];
  int numIPs;
  char allMyIPs[20][20];
  int numCPUs;

  bool sysInfo_first;
  time_t lastSysInfoSend;
 /* The last recorded values for system parameters. */
  double lastSysVals[MAX_SYS_PARAMS];
  /* The current values for the system parameters */
  double currentSysVals[MAX_SYS_PARAMS];
  /* The success/error codes returned by the functions that calculate
     the system parameters */
  int sysRetResults[MAX_SYS_PARAMS];

  /* The current values for the job parameters */
  double currentJobVals[MAX_JOB_PARAMS];
  /* The success/error codes retuprorned by the functions that calculate
     the job parameters */
  int jobRetResults[MAX_JOB_PARAMS];

  /* The current values for the general parameters */
  double currentGenVals[MAX_GEN_PARAMS];
  /* The success/error codes returned by the functions that calculate
     the general parameters */
  int genRetResults[MAX_GEN_PARAMS];

  /* Table which stores the number of processes in each state 
     (R -runnable, S - sleeping etc.) Each entry in the table 
     corresponds to a capital letter. */
  double currentProcessStates[NLETTERS];

  /* CPU information: */
  char cpuVendor[100];
  char cpuFamily[100];
  char cpuModel[100];
  char cpuModelName[200];

  char interfaceNames[20][20];
  int nInterfaces;
  double lastBytesSent[20];
  double lastBytesReceived[20];
  double lastNetErrs[20];
  double *currentNetIn, *currentNetOut, *currentNetErrs;

  double currentNSockets[4];
  double currentSocketsTCP[20];
  char *socketStatesMapTCP[20];  

  /* don't allow a user to send more than MAX_MSG messages per second, in average */
  int maxMsgRate;
  long prvTime;
  double prvSent;
  double prvDrop;
  long crtTime;
  long crtSent;
  long crtDrop;
  double hWeight;

  int instance_id;
  int seq_nr;

 public:
  ApMon(char *initsource) throw(runtime_error);
  

  ApMon(int nDestinations, char **destinationsList) throw(runtime_error);

  ApMon(int nDestinations, char **destAddresses, int *destPorts, char **destPasswds) throw(runtime_error);

  ~ApMon();

  int sendParameter(char *clusterName, char *nodeName,
               char *paramName, int valueType, char *paramValue)
    throw(runtime_error);

  int sendTimedParameter(char *clusterName, char *nodeName,
              char *paramName, int valueType, char *paramValue, int timestamp)
    throw(runtime_error);

  int sendParameter(char *clusterName, char *nodeName,
               char *paramName, int paramValue)
    throw(runtime_error);

  int sendParameter(char *clusterName, char *nodeName,
               char *paramName, float paramValue)
    throw(runtime_error);

  int sendParameter(char *clusterName, char *nodeName,
               char *paramName, double paramValue)
    throw(runtime_error);

  int sendParameter(char *clusterName, char *nodeName,
               char *paramName, char *paramValue)
    throw(runtime_error);


  int sendParameters(char *clusterName, char *nodeName,
               int nParams, char **paramNames, int *valueTypes, 
                         char **paramValues) throw(runtime_error);

  int sendTimedParameters(char *clusterName, char *nodeName,
               int nParams, char **paramNames, int *valueTypes, 
               char **paramValues, int timestamp) throw(runtime_error);

  bool getConfCheck() { return confCheck; }

  long getRecheckInterval() { return recheckInterval; }


  void setRecheckInterval(long val);

  void setConfRecheck(bool confRecheck, long interval);

  void setConfRecheck(bool confRecheck) {
    setConfRecheck(confRecheck, RECHECK_INTERVAL);
  }

  void setJobMonitoring(bool jobMonitoring, long interval);

  void setJobMonitoring(bool jobMonitoring) {
    setJobMonitoring(jobMonitoring, JOB_MONITOR_INTERVAL);
  }

  long getJobMonitorInterval() {
    long i = -1;
    pthread_mutex_lock(&mutexBack);
    if (jobMonitoring)
      i = jobMonitorInterval;
    pthread_mutex_unlock(&mutexBack);
    return i;
  }

  bool getJobMonitoring() {
    bool b;
    pthread_mutex_lock(&mutexBack);
    b = jobMonitoring;
    pthread_mutex_unlock(&mutexBack);
    return b;
  }

  void setSysMonitoring(bool sysMonitoring, long interval); 

  void setSysMonitoring(bool sysMonitoring) {
    setSysMonitoring(sysMonitoring, SYS_MONITOR_INTERVAL);
  }

  long getSysMonitorInterval() {
    long i = -1;
    pthread_mutex_lock(&mutexBack);
    if (sysMonitoring)
      i = sysMonitorInterval;
    pthread_mutex_unlock(&mutexBack);
    return i;
  }

  bool getSysMonitoring() {
    bool b;
    pthread_mutex_lock(&mutexBack);
    b = sysMonitoring;
    pthread_mutex_unlock(&mutexBack);
    return b;
  }

  void setGenMonitoring(bool genMonitoring, int nIntervals);

  void setGenMonitoring(bool genMonitoring) {
    setGenMonitoring(genMonitoring, GEN_MONITOR_INTERVALS);
  }

  bool getGenMonitoring() {
    bool b;
    pthread_mutex_lock(&mutexBack);
    b = genMonitoring;
    pthread_mutex_unlock(&mutexBack);
    return b;
  }

  void addJobToMonitor(long pid, char *workdir, char *clusterName,
                       char *nodeName) throw(runtime_error);

  void removeJobToMonitor(long pid) throw(runtime_error);

  void setSysMonClusterNode(char *clusterName, char *nodeName);

  static void setLogLevel(char *newLevel_s);

  void setMaxMsgRate(int maxRate);

  static void errExit(char *msg);


 protected:

  void initialize(char *filename, bool firstTime) throw(runtime_error);

  void constructFromList(int nDestinations, char **destinationsList)
    throw(runtime_error);

  void initialize(int nDestinations, char **destList, bool firstTime) throw(runtime_error);


  void loadFile(char *filename, int *nDestinations, char **destAddresses,
                int *destPorts, char **destPasswds) throw(runtime_error);

 
  void arrayInit(int nDestinations, char **destAddresses, int *destPorts,
                 char **destPasswds)
    throw(runtime_error);

  void arrayInit(int nDestinations, char **destAddresses, int *destPorts,
                      char **destPasswds, bool firstTime)
    throw(runtime_error);

  void addToDestinations(char *line, int *nDestinations, 
                char *destAddresses[], int destPorts[], char *destPasswds[]); 

  void getDestFromWeb(char *url, int *nDestinations, char *destAddresses[], 
                 int destPorts[], char *destPasswds[],
                      ConfURLs& confURLs) throw(runtime_error);

 
  void encodeParams(int nParams, char **paramNames, int *valueTypes, 
                 char **paramValues, int timestamp) throw(runtime_error);

  void initMonitoring();

  void sendJobInfo();

  void sendOneJobInfo(MonitoredJob job);

  void updateJobInfo(MonitoredJob job); 

  void sendSysInfo();

  void updateSysInfo();

  void sendGeneralInfo();

  void updateGeneralInfo();

  void setBackgroundThread(bool val);

  long getCrtRecheckInterval() {
    return crtRecheckInterval;
  }

  void setCrtRecheckInterval(long val);

  
  void freeConf();

#ifndef WIN32
  friend void *bkTask(void *param);
#else
  friend DWORD WINAPI bkTask(void *param);
#endif
  
  void parseXApMonLine(char *line);

  void initSocket() throw(runtime_error);

  void parseConf(FILE *fp, int *nDestinations, char **destAddresses, 
                     int *destPorts, char **destPasswds)
    throw(runtime_error);

  bool shouldSend();

  friend class ProcUtils;
};

#ifndef WIN32
void *bkTask(void *param); // throw(runtime_error);
#else
DWORD WINAPI bkTask(void *param);
#endif
#endif