source: fact/Evidence/DColl.cc@ 18481

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

  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.
  - The command 'DColl/Log' writes the associated string to the log file
  
  Oliver Grimm, June 2010

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

#define SERVER_NAME "DColl"
#define LOG_FILENAME "Evidence.log"

#include "Evidence.h"

#include <string>
#include <sstream>
#include <iostream>
#include <vector>
#include <sys/stat.h>
#include <regex.h>

using namespace std;

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

        struct Item {
          DimStampedInfo *DataItem;
          bool Exclude;
        };
        vector<struct Item> List;
        
        DimCommand *LogCommand;
        DimInfo *ServerList;
                
    FILE *DataFile;
    FILE *LogFile;
        char *Filename;
        float DataSizeMB, LogSizeMB;
        int DataSizeLastUpdate, LogSizeLastUpdate;
    DimService *LogSizeService, *DataSizeService, *DataFilename, *SubscriptionService;

        int TimeForNextFile;

        vector<regex_t> RegEx;

    void infoHandler();
    void commandHandler();
        void AddService(string);
        void RemoveService(string);
        off_t FileSize(FILE *);

  public:
    DataHandler();
    ~DataHandler();
}; 

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

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

  // Request configuration data / initialise for later non-blocking updates
  GetConfig("basedir");
  GetConfig("sizeupdate");
  GetConfig("rollover");

  // Open log file
  if ((LogFile = fopen((GetConfig("basedir")+"/"+LOG_FILENAME).c_str(), "a")) == NULL) {
    Message(FATAL, "Could not open log file (%s)", strerror(errno));
  }
             
  // Create services for file sizes and data file name
  DataSizeMB = 0;
  DataSizeService = new DimService(SERVER_NAME "/DataSizeMB", DataSizeMB);
  
  LogSizeMB = FileSize(LogFile)/1024.0/1024.0;
  LogSizeService = new DimService(SERVER_NAME "/LogSizeMB", LogSizeMB);

  DataFilename = new DimService(SERVER_NAME "/CurrentFile", (char *) "");

  // Compile regular expressions
  regex_t R;
  vector<string> Exclude = Tokenize(GetConfig("exclude"), " \t");
  for (int i=0; i<Exclude.size(); i++) {
        int Ret = regcomp(&R, Exclude[i].c_str(), REG_EXTENDED|REG_NOSUB);
        if (Ret != 0) {
          char Err[200];
          regerror(Ret, &R, Err, sizeof(Err));
          Message(ERROR, "Error compiling regular expression '%s' (%s)", Exclude[i].c_str(), Err);
        }
        else RegEx.push_back(R);
  }

  // Provide logging command   
  LogCommand = new DimCommand("DColl/Log", (char *) "C", this);

  // Create services for information about subscribed services
  SubscriptionService = new DimService(SERVER_NAME "/Subscriptions", "C", NULL, 0);

  // Subscribe to top-level server list (not via AddService() due to thread issue)
  ServerList = new DimInfo((char *) "DIS_DNS/SERVER_LIST", NO_LINK, this);
}

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

  // Delete all DIM subscriptions
  delete ServerList;
  while (List.size() != 0) RemoveService(List[0].DataItem->getName());
  
  delete LogCommand;
  delete DataFilename;
  delete[] Filename;
  
  delete LogSizeService;
  delete DataSizeService;
  delete SubscriptionService;

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

  // Free memory for regular expressions handling
  for (int i=0; i<RegEx.size(); i++) regfree(&RegEx[i]);
}

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

  static string SubscriptionList;

  // Check if service available
  DimInfo *I = getInfo();
  if (!ServiceOK(I)) return;

  //
  // ====== Part A: Handle service subscriptions ===
  //
  // Services are added here, removal only in destructor.
  
  // If service is DIS_DNS/SERVER_LIST, subscribe to all SERVICE_LIST services
  if (strcmp(I->getName(), "DIS_DNS/SERVER_LIST") == 0) {       
        char *Token = strtok(I->getString(), "+-!@");   
        while (Token != NULL) {
          AddService(string(Token)+"/SERVICE_LIST"); // 'add' also for '-' and '!'
          Token = strtok(NULL, "|"); // Skip server IP address
          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(I->getName(), "/SERVICE_LIST") != NULL) {

        // Bug fix for empty SERVICE_LIST
        if (strlen(I->getString()) == 0) {
          string Tmp(I->getName());
          RemoveService(I->getName());
          AddService(Tmp.c_str());
          return;
        }

        char *Name = strtok(I->getString(), "+-!|");
        while (Name != NULL) {
          // Check if item is a service
          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); // 'add' also for '-' and '!'
          Name = strtok(NULL, "|");
        }
        
        // Update service subscription list
        stringstream Stream;
        
        for (unsigned int i=0; i<List.size(); i++) Stream << List[i].DataItem->getName() << '|';

        SubscriptionList = Stream.str();
        SubscriptionService->updateService((void *) SubscriptionList.c_str(), SubscriptionList.size()+1); // Note that Subscription is a static variable

        return;
  }

  //
  // ====== Part B: Handle opening of data files ===
  //

  // If it is time to open new data file, close the current one 
  if (time(NULL) >= TimeForNextFile) {
        if (DataFile != NULL && fclose(DataFile) != 0) {
          Message(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);

        // Get time structure with date rollover
        if (T->tm_hour >= atoi(GetConfig("rollover").c_str())) T->tm_mday++;
        if (mktime(T) == -1) Message(ERROR, "mktime() failed, check filename");

        // Create direcory if not existing (ignore error if already existing)
        char *Dir;
        if (asprintf(&Dir, "%s/%d", GetConfig("basedir").c_str(), T->tm_year+1900) == -1) {
          Message(FATAL, "asprintf() failed, could not create direcory name");  
        }
        if(mkdir(Dir, S_IRWXU|S_IRWXG)==-1 && errno!=EEXIST) {
          Message(FATAL, "Could not create directory '%s' (%s)", Dir, strerror(errno));
        }

        // Create filename
        free(Filename);
        if (asprintf(&Filename, "%s/%d%02d%02d.slow", Dir, T->tm_year+1900, T->tm_mon+1, T->tm_mday) == 1) {
          Message(FATAL, "asprintf() failed, could not create filename");                 
        }
        free(Dir);
        
        // Open file
        if ((DataFile = fopen(Filename, "a")) == NULL) {
      Message(FATAL, "Could not open data file '%s' (%s)", Filename, strerror(errno));
        }
        else Message(INFO, "Opened data file '%s'", Filename);
        DataFilename->updateService(Filename);
        
        // Calculate time for next file opening
        T->tm_sec = 0;
        T->tm_min = 0;
        T->tm_hour = atoi(GetConfig("rollover").c_str());
        TimeForNextFile = mktime(T);
  }

  //
  // ====== Part C: Handle writing to data file ===
  //

  // Identify index of service
  for (int Service=0; Service<List.size(); Service++) if (I == List[Service].DataItem) {

        // Service excluded from writing or contains no data?
        if (List[Service].Exclude || I->getSize()==0) return;
        
        // Write data header
        time_t RawTime = I->getTimestamp();
        struct tm *TM = localtime(&RawTime);

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

        // Translate data into ASCII
        string Text = EvidenceServer::ToString(I->getFormat(), I->getData(), I->getSize());

        if (!Text.empty()) {
          // Replace new line by '\' and all other control characters by white space
          for (int i=0; i<Text.size(); i++) {
                if (Text[i] == '\n') Text[i] = '\\';
                else if (iscntrl(Text[i])) Text[i] = ' ';
          }
          // Write to file
      fprintf(DataFile, "%s\n", Text.c_str());
        }
        else fprintf(DataFile, "Cannot interpret service format\n");

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

        // Update datafile size service (not every time to avoid infinite loop)
        if (time(NULL) - DataSizeLastUpdate > max(atoi(GetConfig("sizeupdate").c_str()), 1)) {
      fflush(DataFile); // not continuously to reduce load

          DataSizeMB = FileSize(DataFile)/1024.0/1024.0;
          DataSizeService->updateService();
          DataSizeLastUpdate = time(NULL);
        }  
  } // Check for disk writing
}

//
// Implementation of log writing
//
void DataHandler::commandHandler() {

  // Translate text safely to string
  string Text = ToString((char *) "C", getCommand()->getData(), getCommand()->getSize());

  // Safety check
  if (getCommand() != LogCommand || LogFile == NULL || Text.empty()) return;
  
  // Replace all carriage-return by line feed and non-printable characters
  for (unsigned int i=0; i<Text.size(); 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, "%.2d/%.2d/%4d %.2d:%.2d:%2.d %s (ID %d): %s\n",
                TM->tm_mday, TM->tm_mon+1, TM->tm_year+1900,
                TM->tm_hour, TM->tm_min, TM->tm_sec, getClientName(), getClientId(), Text.c_str());
  fflush(LogFile);
  
  // If error close file (otherwise infinite loop because Message() also writes to log) 
  if(ferror(LogFile)) {
    fclose(LogFile);
        LogFile = NULL;
    Message(ERROR, "Error writing to log file, closing file (%s)", strerror(errno));
  }
    
  // Update logfile size service (not every time to avoid infinite loop)
  if (time(NULL) - LogSizeLastUpdate > atoi(GetConfig("sizeupdate").c_str())) {
        LogSizeMB = FileSize(LogFile)/1024.0/1024.0;
        LogSizeService->updateService();
        LogSizeLastUpdate = time(NULL);
  }
}


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

  struct Item New;

  // Check if already subscribed to this service
  for (int i=0; i<List.size(); i++) {
        if (Name == List[i].DataItem->getName()) return;
  }
  
  // Should service be ignored?
  New.Exclude = false;
  for (int i=0; i<RegEx.size(); i++) {
    if (regexec(&RegEx[i], Name.c_str(), (size_t) 0, NULL, 0) == 0) New.Exclude = true;
  }

  // Subscribe to service
  New.DataItem = new DimStampedInfo(Name.c_str(), NO_LINK, this);

  List.push_back(New);
}


//
// Remove service from watch list
//
void DataHandler::RemoveService(string Name) {

  // Find service index
  vector<struct Item>::iterator E;
  for (E=List.begin(); E<List.end(); ++E) if (Name == (*E).DataItem->getName()) {
        delete (*E).DataItem;   
        List.erase(E);
  }
}

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

  struct stat FileStatus;

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


//          
// Main program
//
int main() {

  dic_disable_padding();
  dis_disable_padding();
        
  // Static ensures calling of destructor by exit()
  static DataHandler DataInstance;
  
  // Sleep until exit requested
  while (!DataInstance.ExitRequest) pause();
}