source: Evidence/DColl.cc @ 145

/********************************************************************\

  Central data collector of the Evidence Control System
  
  - DColl subscribes to all services given by the configuration
    server and writes these to the data file at every update.
  - One data file per day is generated, with roll-over at 13:00 local time.
  - For each service, it creates a new service with '.hist' appended
        that contains a history of events kept within a ring buffer. Each
        entry will be the result of a conversion to double of the text 
        written to the data file. Only if the new value has changed by a
        minimum amout it will be added to the ring buffer.
  - The command 'DColl/Log' writes the associated string to the log 
    file specified in the configuration.
  
  Oliver Grimm, December 2009

\********************************************************************/

#define SERVER_NAME "DColl"

#define DATE_ROLLOVER 12 // localtime hour after which next date is used

#include "Evidence.h"

#include <math.h>
#include <float.h>
#include <sys/stat.h>
#include <ctype.h>
#include <sys/types.h>
#include <regex.h>

//
// Class declaration
//
class DataHandler:      public DimClient, public DimBrowser,
                                        public EvidenceServer {

        struct Item {
          DimStampedInfo *DataItem;
      DimService *HistService;
          struct EvidenceHistoryItem *HistBuffer;
          int HistPointer;
          double LastValue;
          double MinAbsChange;
        } *List;

        DimCommand *LogCommand;
                
    unsigned int NumItems;
    FILE *DataFile;
    FILE *LogFile;
        float DataSizekB, LogSizekB;
        int DataSizeLastUpdate, LogSizeLastUpdate;
        char *DataDir;
    DimService *LogSizeService, *DataSizeService;
    int HistSize;
        int SizeUpdateDelay;
        int TimeForNextFile;
        
        int RegExCount;
        regex_t *RegEx;
        double *MinChange;
        
    void infoHandler();
    void commandHandler();
        void AddService(char *);
        float FileSize(FILE *);
            
  public:
    DataHandler();
    ~DataHandler();
}; 

//
// Constructor
//
DataHandler::DataHandler(): EvidenceServer(SERVER_NAME) {

  // Initialization to prevent freeing unallocated memory 
  DataFile = NULL;
  LogFile = NULL;
  List = NULL;
  LogSizeService = NULL;
  DataSizeService = NULL;
  
  DataSizeLastUpdate = 0;
  LogSizeLastUpdate = 0;
  TimeForNextFile = 0;

  // Request configuration data
  char *Change = GetConfig(SERVER_NAME " minchange");
  DataDir = GetConfig(SERVER_NAME " datadir");
  char *Logname = GetConfig(SERVER_NAME " logfile");
  SizeUpdateDelay = atoi(GetConfig(SERVER_NAME " sizeupdate"));
  HistSize = atoi(GetConfig(SERVER_NAME " histsize"));
  if (HistSize < 1) HistSize = 1; // Minimum one items
   
  // Open log file
  if ((LogFile = fopen(Logname, "a")) == NULL) {
    State(FATAL, "Could not open log file '%s' (%s)", Logname, strerror(errno));
  }

  // Provide logging command   
  LogCommand = new DimCommand("DColl/Log", (char *) "C", this);
              
  // Create services for file sizes
  DataSizekB = 0;
  DataSizeService = new DimService(SERVER_NAME "/DataSizekB", DataSizekB);
  
  LogSizekB = FileSize(LogFile);
  LogSizeService = new DimService(SERVER_NAME "/LogSizekB", LogSizekB);

  // Count how many minimum change value regular expressions are present
  RegExCount = 0;
  char *Token = strtok(Change, "\t ");
  while (Token != NULL) {
        RegExCount++;
        Token = strtok(NULL, "\t ");
  }
  
  // Allocate memory for regular expressions and minimum change values
  RegEx = new regex_t[RegExCount];
  MinChange = new double [RegExCount];

  // Compile regular expressions
  int Pos = 0;
  for (int i=0; i<RegExCount; i++) {
    int Len = strlen(Change+Pos) + 1;
    Token = strtok(Change + Pos, ": \t");

    int Ret = regcomp(&RegEx[i], Token, REG_EXTENDED|REG_NOSUB);
        if (Ret != 0) {
          char ErrMsg[200];
          regerror(Ret, &RegEx[i], ErrMsg, sizeof(ErrMsg));
          State(WARN, "Error compiling regular expression '%s' (%s)", Token, ErrMsg);
        }
        else {
          if ((Token=strtok(NULL, "")) != NULL) MinChange[i] = atof(Token);
          else MinChange[i] = 0;
        }
        Pos += Len;
  }

  // Subscribe to list of servers at DIS_DNS 
  AddService((char *) "DIS_DNS/SERVER_LIST");

  DimClient::sendCommand("DColl/Log", SERVER_NAME" *** Logging started ***");
}

//
// Destructor: Close files and free memory
//
DataHandler::~DataHandler() {

  // Delete DIM services and command first so handlers and not called anymore
  for (int i=0; i<NumItems; i++) {
        delete List[i].HistService;
    delete List[i].DataItem;
    delete[] List[i].HistBuffer;
  }
  free(List);

  //delete LogSizeService; // These create segmentation faults?!
  //delete DataSizeService;

  delete LogCommand;

  // Close files
  if (LogFile != NULL) if (fclose(LogFile) != 0) {
        State(ERROR, "Could not close log file (%s)", strerror(errno));
  }
  if (DataFile != NULL && fclose(DataFile) != 0) {
        State(ERROR, "Error: Could not close data file (%s)", strerror(errno));
  }

  // Free memory for regular expressions handling
  for (int i=0; i<RegExCount; i++) {
    regfree(&RegEx[i]);
  }
  delete[] MinChange;
  delete[] RegEx;
}

//
// Implementation of data handling
//
// DIM ensures infoHandler() is called serialized, therefore
// no mutex is needed to serialize writing to the file
void DataHandler::infoHandler() {

  DimInfo *Info = getInfo();

  // Check if service available
  if (Info->getSize()==strlen(NO_LINK)+1 && strcmp(Info->getString(), NO_LINK)==0) return;

  // If service is DIS_DNS/SERVER_LIST, subscribe to all SERVICE_LIST services
  if (strcmp(Info->getName(), "DIS_DNS/SERVER_LIST") == 0) {    
        char *Token = strtok(Info->getString(), "+-!|@");       
        while (Token != NULL) {
          char *Buf;
          if (MakeString(&Buf, "%s/SERVICE_LIST", Token) != -1) {
            AddService(Buf);
                free(Buf);
          }
          else State(ERROR, "MakeString() failed for server %s", Token);
          
          Token = strtok(NULL, "|");
          Token = strtok(NULL, "+-!|@");        
        }       
        return;
  }

  // If service is SERVICE_LIST of any server, scan all services.
  // Subscribe to all services (but not to commands and RPCs)
  if (strstr(Info->getName(), "/SERVICE_LIST") != NULL) {
        char *Name = strtok(Info->getString(), "+-!|");
        while (Name != NULL) {
          char *Type = strtok(NULL, "\n");
          if (Type == NULL) return; // for safety, should not happen
      if (strstr(Type, "|CMD")==NULL && strstr(Type, "|RPC")==NULL) {
                AddService(Name);
          }
          Name = strtok(NULL, "+-!|");
        }
        return;
  }

  // If it is time to open new data file, close the current one 
  if (time(NULL) >= TimeForNextFile) {
        if (DataFile != NULL && fclose(DataFile) != 0) {
          State(ERROR, "Error: Could not close data file (%s)", strerror(errno));
        }
        DataFile = NULL;
  }
  
  // Open new data file if necessary
  if (DataFile == NULL) {
        time_t Time = time(NULL);
        struct tm *T = localtime(&Time);
        
        if(T->tm_hour >= DATE_ROLLOVER) T->tm_mday++;
        if (mktime(T) == -1) State(ERROR, "mktime() failed, check filename");

        char *Filename;
        if (MakeString(&Filename, "%s/%d%02d%02d.slow", DataDir, T->tm_year+1900, T->tm_mon+1, T->tm_mday) == -1) State(FATAL, "Could not create filename, MakeString() failed");
        if ((DataFile = fopen(Filename, "a")) == NULL) {
      State(FATAL, "Could not open data file '%s' (%s)", Filename, strerror(errno));
        }
        else State(INFO, "Opened data file '%s'", Filename);
        free(Filename);
        
        // Calculate time for next file opening
        T->tm_sec = 0;
        T->tm_min = 0;
        T->tm_hour = DATE_ROLLOVER;
        TimeForNextFile = mktime(T);
  }
   
  // Identify index of service
  int Service;  
  for (Service=0; Service<NumItems; Service++) if (Info == List[Service].DataItem) break;
  if (Service == NumItems) return;  // Service not found

  // If negative value for absolute change, ignore this entry
  if (List[Service].MinAbsChange < 0) return;

  // Write data header
  time_t RawTime = Info->getTimestamp();
  struct tm *TM = localtime(&RawTime);

  fprintf(DataFile, "%s %d %d %d %d %d %d %d %lu ", Info->getName(), TM->tm_year+1900, TM->tm_mon+1, TM->tm_mday, TM->tm_hour, TM->tm_min, TM->tm_sec, Info->getTimestampMillisecs(), Info->getTimestamp());

  // Translate data into ASCII
  char *Text = EvidenceServer::ToString(Info);
  if (Text != NULL) {
        // Replace all control characters by white space
        for (int i=0; i<strlen(Text); i++) if (iscntrl(Text[i])) Text[i] = ' ';
        
        // Write to file
    fprintf(DataFile, "%s\n", Text);
        
        // Add to history buffer if change large enough
        if ((fabs(atof(Text)-List[Service].LastValue) > List[Service].MinAbsChange)) {
          List[Service].HistBuffer[List[Service].HistPointer].Seconds = Info->getTimestamp();
      List[Service].HistBuffer[List[Service].HistPointer].Value = atof(Text);
      List[Service].HistService->updateService();
          List[Service].HistPointer++;
          if (List[Service].HistPointer >= HistSize) List[Service].HistPointer = 0;
          List[Service].LastValue = atof(Text);
        }
        free(Text);
  }
  else fprintf(DataFile, "Cannot interpret format identifier\n");
 
  fflush(DataFile);

  // Terminate if error because otherwise infinite loop might result as
  // next call to this infoHandler() will try to (re-)open file
  if(ferror(DataFile)) {
    fclose(DataFile);
        DataFile = NULL;
        State(FATAL, "Error writing to data file, closed file (%s)", strerror(errno));
  }

  // Update datafile size service
  if (time(NULL) - DataSizeLastUpdate > SizeUpdateDelay) {
        DataSizekB = FileSize(DataFile);
        DataSizeService->updateService();
        DataSizeLastUpdate = time(NULL);
  }
}

//
// Implementation of log writing
//
void DataHandler::commandHandler() {
 
  if (getCommand() != LogCommand || LogFile == NULL) return;

  // Replace all carriage-return by line feed and non-printable characters
  char *Text = getCommand()->getString();
  for (unsigned int i=0; i<strlen(Text); i++) {
    if (Text[i] == '\r') Text[i] = '\n';
        if(isprint(Text[i])==0 && isspace(Text[i])==0) Text[i] = ' ';
  }
 
  time_t RawTime = time(NULL);
  struct tm *TM = localtime(&RawTime);

  fprintf(LogFile, "%d/%d/%d %d:%d:%d: %s\n",
                TM->tm_mday, TM->tm_mon+1, TM->tm_year+1900,
                TM->tm_hour, TM->tm_min, TM->tm_sec, Text);

  fflush(LogFile);
  
  // If error close file (otherwise infinite loop because State() also writes to log) 
  if(ferror(LogFile)) {
    fclose(LogFile);
        LogFile = NULL;
    State(ERROR, "Error writing to log file, closing file (%s)", strerror(errno));
  }
    
  // Update logfile size service
  if (time(NULL) - LogSizeLastUpdate > SizeUpdateDelay) {
        LogSizekB = FileSize(LogFile);
        LogSizeService->updateService();
        LogSizeLastUpdate = time(NULL);
  }
}


//
// Add service to watch list
//
void DataHandler::AddService(char *Name) {

  // Do not subscribe to history services (otherwise infinite loop)
  if (strstr(Name, ".hist") != NULL) return;

  // Check if already subscribed to this service
  for (int i=0; i<NumItems; i++) {
        if(strcmp(Name, List[i].DataItem->getName()) == 0) return;
  }

  // Increase capacity of item list                     
  struct Item *New = (struct Item *) realloc(List, (NumItems+1)*sizeof(struct Item));
  if (New != NULL) List = New;
  else {
        State(ERROR, "Could not allocate memory for item list, service '' not added (%s)", Name, strerror(errno));
        return;
  }
  
  // Set minimum required change by comparing to regular expressions
  List[NumItems].MinAbsChange = 0;
  for (int i=0; i<RegExCount; i++) {
    if (regexec(&RegEx[i], Name, (size_t) 0, NULL, 0) == 0) {
          List[NumItems].MinAbsChange = MinChange[i];
        }
  }
  
  List[NumItems].LastValue = DBL_MAX;
                
  // Create history service
  List[NumItems].HistBuffer = new struct EvidenceHistoryItem [HistSize];
  memset(List[NumItems].HistBuffer, 0, HistSize*sizeof(EvidenceHistoryItem));
  List[NumItems].HistPointer = 0;

  char *Buf;
  if (MakeString(&Buf, "%s.hist", Name) == -1) {
    State(ERROR, "Could not create history service for '%s', MakeString() failed", Name);
  }
  else {
        List[NumItems].HistService = new DimService (Buf, (char *) "C",
                                          List[NumItems].HistBuffer, HistSize*sizeof(EvidenceHistoryItem));
    free(Buf);
  }

  // Subscribe to service
  List[NumItems].DataItem = new DimStampedInfo(Name, NO_LINK, this);

  // Increase number only after all set-up
  NumItems++;
}

//
// Determine size of file in kB
//
float DataHandler::FileSize(FILE *File) {

  struct stat FileStatus;

  if (fstat(fileno(File), &FileStatus) == -1) {
     State(WARN, "Could not determine size of file (%s)", strerror(errno));
         return -1;
  }

  return (float) FileStatus.st_size/1024;
}

//          
// Main program
//
int main() {
        
  // Static ensures calling of destructor by exit()
  static DataHandler Data;
  
  // Sleep until signal caught
  pause();
}