Index: Evidence/DColl/DColl.cc
===================================================================
--- Evidence/DColl/DColl.cc	(revision 128)
+++ Evidence/DColl/DColl.cc	(revision 142)
@@ -5,7 +5,10 @@
   - DColl subscribes to all services given by the configuration
     server and writes these to the data file at every update.
-  - 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.
+  - 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.
@@ -17,31 +20,50 @@
 #define SERVER_NAME "DColl"
 
+#define DATE_ROLLOVER 12 // localtime hour after which next date is used
+
 #include "../Evidence.h"
-
-#define NO_LINK "__&DIM&NOLINK&__" // for checking if DIMserver is alive
+#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 DimCommand,
-					public DimBrowser, public EvidenceServer {
-  private:
-    pthread_mutex_t DataMutex;
-    pthread_mutex_t LogMutex;
-
+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;
-    char *Filename;
-
-    DimStampedInfo **DataItem;
-    DimService **HistService;	    
+	float DataSizekB, LogSizekB;
+	int DataSizeLastUpdate, LogSizeLastUpdate;
+	char *DataDir;
+    DimService *LogSizeService, *DataSizeService;
     int HistSize;
-    struct EvidenceHistoryItem **HistBuffer;
-	int *HistPointer;
-    	
+	int SizeUpdateDelay;
+	int TimeForNextFile;
+	
+	int RegExCount;
+	regex_t *RegEx;
+	double *MinChange;
+	
     void infoHandler();
     void commandHandler();
-    bool ReallocMem(void *&, int);
+	void AddService(char *);
+	float FileSize(FILE *);
 	    
   public:
@@ -53,92 +75,75 @@
 // Constructor
 //
-DataHandler::DataHandler(): DimCommand((char *) "DColl/Log", (char *) "C"), EvidenceServer(SERVER_NAME) {
+DataHandler::DataHandler(): EvidenceServer(SERVER_NAME) {
 
   // Initialization to prevent freeing unallocated memory 
   DataFile = NULL;
-  Filename = NULL;
-
-  DataItem = NULL;
-  HistService = NULL;
-  HistBuffer = NULL;
-  HistPointer = NULL;
-  
+  LogFile = NULL;
+  List = NULL;
+  LogSizeService = NULL;
+  DataSizeService = NULL;
+  
+  DataSizeLastUpdate = 0;
+  LogSizeLastUpdate = 0;
+  TimeForNextFile = 0;
+
   // Request configuration data
-  char *Items = GetConfig(SERVER_NAME " items");
-  char *DataDir = GetConfig(SERVER_NAME " datadir");
+  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
    
-  // Create mutex for thread synchronization 
-  if ((errno=pthread_mutex_init(&DataMutex, NULL)) != 0) {
-    Msg(FATAL, "pthread_mutex_init() failed for data file (%s)", strerror(errno));
-  }
-  if ((errno=pthread_mutex_init(&LogMutex, NULL)) != 0) {
-    Msg(FATAL, "pthread_mutex_init() failed for log file (%s)", strerror(errno));
-  }
-
-  // Interpret DIM services to observe and prepare history buffer
-  char *Buf, *ServiceName, *Format;
-  int NumServices;
-  NumItems = 0;
-  char *NextToken = strtok(Items, " \t");
-  while (NextToken != NULL) {
-    NumServices = getServices(NextToken);
-    for (int j=0; j<NumServices; j++)  {
-	  if (getNextService(ServiceName, Format) == DimSERVICE) {
-
-	    // Check if already subsubed to this service
-		for (int i=0; i<NumItems; i++) {
-		  if(strcmp(ServiceName, DataItem[i]->getName()) == 0) continue;
-		}
-		
-	    // Increase capacity of arrays
-	    if (!ReallocMem((void *&) DataItem, NumItems+1)) continue;
-	    if (!ReallocMem((void *&) HistService, NumItems+1)) continue;
-	    if (!ReallocMem((void *&) HistBuffer, NumItems+1)) continue;
-	    if (!ReallocMem((void *&) HistPointer, NumItems+1)) continue;
-
-  		// Subscribe to service
-    	DataItem[NumItems] = new DimStampedInfo(ServiceName, (char *) NO_LINK, this);
-
-    	// Create history service
-    	HistBuffer[NumItems] = new struct EvidenceHistoryItem [HistSize];
-		memset(HistBuffer[NumItems], 0, HistSize*sizeof(EvidenceHistoryItem));
-		HistPointer[NumItems] = 0;
-
-    	if (asprintf(&Buf, "%s.hist", ServiceName) == -1) {
-    	  Msg(ERROR, "Could not create Hist service for %s because asprintf() failed", ServiceName);
-    	}
-    	else {
-		  HistService[NumItems] = new DimService (Buf, (char *) "C",
-	  							HistBuffer[NumItems], HistSize*sizeof(EvidenceHistoryItem));
-    	  free(Buf);
-		}
-		NumItems++;
-	  }
-	}
-    NextToken = strtok(NULL, " \t");
-  }
-
-  // Open data file
-  time_t rawtime = time(NULL);
-  struct tm * timeinfo = gmtime(&rawtime);
-  if(timeinfo->tm_hour >= 13) rawtime += 12*60*60;
-  timeinfo = gmtime(&rawtime);
-
-  if (asprintf(&Filename, "%s/%d%02d%02d.slow", DataDir, timeinfo->tm_year+1900, timeinfo->tm_mon+1, timeinfo->tm_mday) == -1) {
-    Filename = NULL;
-    Msg(FATAL, "Could not create filename with asprintf()(%s)", strerror(errno));
-  }
-  if ((DataFile = fopen(Filename, "a")) == NULL) {
-    Msg(FATAL, "Could not open data file '%s' (%s)", Filename, strerror(errno));
-  }
-  
   // Open log file
   if ((LogFile = fopen(Logname, "a")) == NULL) {
-    Msg(FATAL, "Could not open log file '%s' (%s)", Logname, strerror(errno));
-  }
-
+    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 ***");
 }
 
@@ -148,27 +153,31 @@
 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 && fclose(LogFile) != 0) {
-	Msg(ERROR, "Could not close log file (%s)", strerror(errno));
+  if (LogFile != NULL) if (fclose(LogFile) != 0) {
+	State(ERROR, "Could not close log file (%s)", strerror(errno));
   }
   if (DataFile != NULL && fclose(DataFile) != 0) {
-	Msg(ERROR, "Error: Could not close data file (%s)", strerror(errno));
-  }	
-
-  // Free all memory
-  for (int i=0; i<NumItems; i++) {
-    delete[] HistBuffer[i];
-    delete DataItem[i];
-  }
-
-  free(Filename);
-  free(HistService);
-  free(HistPointer);
-  free(HistBuffer);
-  free(DataItem);
-  
-  // Destroy mutex
-  pthread_mutex_destroy (&LogMutex);
-  pthread_mutex_destroy (&DataMutex);
+	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;
 }
 
@@ -176,21 +185,84 @@
 // Implementation of data handling
 //
-// More than one infoHandler() might run in parallel, therefore
-// the mutex mechanism is used to serialize writing to the file
+// 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 actually available, if it contains data and if data file is open
+  // Check if service available
   if (Info->getSize()==strlen(NO_LINK)+1 && strcmp(Info->getString(), NO_LINK)==0) return;
-  if (Info->getSize() == 0 || DataFile == NULL) 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 == DataItem[Service]) break;
-  if (Service == NumItems) return;  // Not found: should never happen
-
-  pthread_mutex_lock(&DataMutex);
-  
+  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();
@@ -199,15 +271,22 @@
   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 info data into ASCII and write to file and to history buffer
+  // 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);
 	
-	HistBuffer[Service][HistPointer[Service]].Seconds = Info->getTimestamp();
-    HistBuffer[Service][HistPointer[Service]].Value = atof(Text);
-    HistService[Service]->updateService();
-	HistPointer[Service]++;
-	if (HistPointer[Service] >= HistSize) HistPointer[Service] = 0;
-
+	// 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);
   }
@@ -215,18 +294,35 @@
  
   fflush(DataFile);
-  if(ferror(DataFile)) Msg(ERROR, "Error writing to data file (%s)", strerror(errno));
-
-  pthread_mutex_unlock(&DataMutex);
-}
-
-//
-// Implementation of log writing (the only command is 'DColl/Log')
+
+  // 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 (LogFile == NULL) return;  // Handler might be called before file is opened
  
-  pthread_mutex_lock(&LogMutex);
-
+  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);
@@ -234,27 +330,90 @@
   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, getString());
+		TM->tm_hour, TM->tm_min, TM->tm_sec, Text);
 
   fflush(LogFile);
-  if(ferror(LogFile)) Msg(ERROR, "Error writing to log file (%s)", strerror(errno));
-  pthread_mutex_unlock(&LogMutex);
-}
-
-
-//
-// Implementation of memory re-allocation for pointer arrays
-//
-bool DataHandler::ReallocMem(void *&Memory, int Size) {
-
-  void *NewMem = realloc(Memory, Size*sizeof(void *));
-
-  if (NewMem != NULL) {
-    Memory = NewMem;
-	return true;
-  }
+  
+  // 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 {
-	Msg(ERROR, "Could not allocate memory ()", strerror(errno));
-	return false;
-  }
+	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;
 }
 
@@ -267,4 +426,5 @@
   static DataHandler Data;
   
-  pause();	  // Sleep until signal caught
-}
+  // Sleep until signal caught
+  pause();
+}