//**************************************************************** /** @class DataLogger @brief Logs all message and infos between the services This is the main logging class facility. It derives from StateMachineDim and DimInfoHandler. the first parent is here to enforce a state machine behaviour, while the second one is meant to make the dataLogger receive dim services to which it subscribed from. The possible states and transitions of the machine are: \dot digraph datalogger { node [shape=record, fontname=Helvetica, fontsize=10]; e [label="Error" color="red"]; r [label="Ready"] d [label="NightlyOpen"] w [label="WaitingRun"] l [label="Logging"] b [label="BadNightlyconfig" color="red"] c [label="BadRunConfig" color="red"] e -> r r -> e r -> d r -> b d -> w d -> r w -> r l -> r l -> w b -> d w -> c w -> l b -> r c -> r c -> l } \enddot For questions or bug report, please contact Etienne Lyard (etienne.lyard@unige.ch) or Thomas Bretz. */ //**************************************************************** #include //for getting stat of opened files //#include //for getting disk free space //#include //for getting files sizes #include #include #include #include "Dim.h" #include "Event.h" #include "StateMachineDim.h" #include "Configuration.h" #include "Converter.h" #include "DimWriteStatistics.h" #include "Description.h" #include "DimNetwork.h" #ifdef HAVE_FITS #include "Fits.h" #endif //Dim structures ///distributes the number of opened subscriptions and fits files struct NumSubAndFitsType { int numSubscriptions; int numOpenFits; }; ///distributes which files were opened. struct OpenFileToDim { int code; char fileName[FILENAME_MAX]; }; ///Run number record. Used to keep track of which run numbers are still active struct RunNumberType { #ifdef RUN_LOGS ///the run number log file shared_ptr logFile; #endif ///the run number report file shared_ptr reportFile; #ifdef HAVE_FITS ///the run number group fits file shared_ptr runFitsFile; #endif #ifdef RUN_LOGS ///the log filename string logName; #endif ///the report filename string reportName; ///the actual run number uint32_t runNumber; ///the time at which the run number was received Time time; ///list of opened fits used to create the fits grouping when the run ends map > openedFits; ///default constructor RunNumberType() { #ifdef RUN_LOGS logFile = shared_ptr(new ofstream()); #endif reportFile = shared_ptr(new ofstream()); #ifdef HAVE_FITS runFitsFile = shared_ptr(); #endif runNumber = 0; } ///default destructor ~RunNumberType() { } void addServiceToOpenedFits(const string& fileName, const string& serviceName) { //most likely I should add this service name. //the only case for which I should not add it is if a service disapeared, hence the file was closed //and reopened again. Unlikely to happen, but well it may if (find(openedFits[fileName].begin(), openedFits[fileName].end(), serviceName)==openedFits[fileName].end()) openedFits[fileName].push_back(serviceName); } }; ///Dim subscription type. Stores all the relevant info to handle a Dim subscription struct SubscriptionType { #ifdef HAVE_FITS ///Nightly FITS output file Fits nightlyFile; ///run-specific FITS output file Fits runFile; #endif ///the actual dimInfo pointer shared_ptr dimInfo; ///the server string server; ///the service string service; ///the converter for outputting the data according to the format shared_ptr fConv; ///the current run number used by this subscription uint32_t runNumber; ///time of the latest received event Time lastReceivedEvent; ///whether or not the fits buffer was allocated already bool fitsBufferAllocated; ///Dim info constructor SubscriptionType(DimStampedInfo* info=NULL) { dimInfo = shared_ptr(info); fConv = shared_ptr(); runNumber = 0; lastReceivedEvent = Time::None; fitsBufferAllocated = false; } ///default destructor ~SubscriptionType() { } }; class DataLogger : public StateMachineDim, DimServiceInfoList { public: /// The list of existing states specific to the DataLogger enum { kSM_NightlyOpen = 20, ///< Nightly file openned and writing kSM_WaitingRun = 30, ///< waiting for the run number to open the run file kSM_Logging = 40, ///< both files openned and writing kSM_BadNightlyConfig = 0x101, ///< the folder specified for Nightly logging does not exist or has bad permissions kSM_BadRunConfig = 0x102, ///< the folder specified for the run logging does not exist or has wrong permissions or no run number kSM_RunWriteError = 0x103, ///< Denotes that an error occured while writing a run file (text or fits). kSM_DailyWriteError = 0x103,///< Denots that an error occured while writing a daily file (text or fits). } localstates_t; DataLogger(ostream &out); ~DataLogger(); int EvalConfiguration(const Configuration& conf); private: /************************************************ * MEMBER VARIABLES ************************************************/ /// ofstream for the NightlyLogfile ofstream fNightlyLogFile; /// ofstream for the Nightly report file ofstream fNightlyReportFile; /// base path of the Nightlyfile string fNightlyFilePath; ///base path of the run file string fRunFilePath; ///run numbers list fRunNumber; ///old run numbers time-out delay (in seconds) uint32_t fRunNumberTimeout; ///previous run number. to check if changed while logging int fPreviousRunNumber; ///Current Service Quality int fQuality; ///Modified Julian Date double fMjD; ///for obtaining the name of the existing services // ServiceList fServiceList; typedef map > SubscriptionsListType; ///All the services to which we have subscribed to, sorted by server name. SubscriptionsListType fServiceSubscriptions; ///full name of the nightly log file string fFullNightlyLogFileName; ///full name of the nightly report file string fFullNightlyReportFileName; ///variable to track when the statistic were last calculated // Time fPreviousStatsUpdateTime; Time fPreviousOldRunNumberCheck; ///boolean to know whether we should close and reopen daily files or not bool fDailyFileDayChangedAlready; DimWriteStatistics fFilesStats; private: /*************************************************** * DIM INFO HANDLER ***************************************************/ //overloading of DIM's infoHandler function void infoHandler(); /*************************************************** * TRANSITION FUNCTIONS ***************************************************/ ///Reporting method for the services info received void ReportPlease(DimInfo* I, SubscriptionType& sub); int ConfigureFileName(string &target, const string &type, const EventImp &evt); ///Configuration of the nightly file path int ConfigureNightlyFileName(const Event& evt); ///Configuration fo the file name int ConfigureRunFileName(const Event& evt); ///DEPREC - configuration of the run number int ConfigureRunNumber(const Event& evt); ///print the current state of the dataLogger int PrintStatePlease(const Event& evt); ///checks whether or not the current info being treated is a run number void CheckForRunNumber(DimInfo* I); /// start transition int StartPlease(); ///from waiting to logging transition int StartRunPlease(); /// from logging to waiting transition int StopRunPlease(); ///stop and reset transition int GoToReadyPlease(); ///from NightlyOpen to waiting transition int NightlyToWaitRunPlease(); ///from wait for run number to nightly open int BackToNightlyOpenPlease(); #ifdef HAVE_FITS ///Open fits files void OpenFITSFilesPlease(SubscriptionType& sub, RunNumberType* cRunNumber); ///Write data to FITS files void WriteToFITS(SubscriptionType& sub); ///Allocate the buffers required for fits void AllocateFITSBuffers(SubscriptionType& sub); #endif//has_fits /*************************************** * DIM SERVICES PROVIDED BY THE DATA LOGGER ***************************************/ ///monitoring notification loop void ServicesMonitoring(); inline void NotifyOpenedFile(const string &name, int type, DimDescribedService* service); ///Service for opened files DimDescribedService* fOpenedNightlyFiles; DimDescribedService* fOpenedRunFiles; DimDescribedService* fNumSubAndFits; NumSubAndFitsType fNumSubAndFitsData; /*************************************************** * DATA LOGGER's CONFIGURATION STUFF ***************************************************/ ///black/white listing set fBlackList; set fWhiteList; ///list of services to be grouped set fGrouping; ///configuration flags bool fDebugIsOn; // float fStatsPeriodDuration; bool fOpenedFilesIsOn; bool fNumSubAndFitsIsOn; //functions for controlling the services behavior int SetDebugOnOff(const Event& evt); int SetStatsPeriod(const Event& evt); int SetOpenedFilesOnOff(const Event& evt); int SetNumSubsAndFitsOnOff(const Event& evt); int SetRunTimeoutDelay(const Event& evt); ///boolean to prevent DIM update while desctructing the dataLogger bool fDestructing; /*************************************************** * UTILITIES ***************************************************/ ///vectors to keep track of opened Fits files, for grouping purposes. map > fOpenedNightlyFits; ///creates a group fits file based on a list of files to be grouped void CreateFitsGrouping(map >& filesToGroup, int runNumber); bool OpenStream(shared_ptr stream, const string &filename); ///Open the relevant text files related to a particular run int OpenRunFile(RunNumberType& run); ///add a new run number void AddNewRunNumber(int64_t newRun, Time time); std::vector previousRunNumbers; ///removes the oldest run number, and close the relevant files. void RemoveOldestRunNumber(); ///retrieves the size of a file off_t GetFileSize(const string&); ///Get the digits of year, month and day for filenames and paths void GetYearMonthDayForFiles(unsigned short& year, unsigned short& month, unsigned short& day); ///Appends the relevant year month day to a given path void AppendYearMonthDaytoPath(string& path); ///Form the files path string CompileFileNameWithPath(const string &path, const string &service, const string & extension, const Time &time=Time()); ///Form the file names only string CompileFileName(const string& service, const string& extension, const Time& time=Time()); ///Form the files path string CompileFileNameWithPath(const string &path, const uint32_t run, const string &service, const string & extension, const Time &time=Time()); ///Form the file names only string CompileFileName(const uint32_t run, const string& service, const string& extension);//, const Time& time=Time()); ///Check whether service is in black and/or white list bool ShouldSubscribe(const string& server, const string& service); ///Subscribe to a given server and service DimStampedInfo* SubscribeToPlease(const string& server, const string& service); ///Open a text file and checks for ofstream status bool OpenTextFilePlease(ofstream& stream, const string& name); ///Checks if the input osftream is in error state, and if so close it. bool CheckForOfstreamError(ofstream& out, bool isDailyStream); ///Goes to Write error states void GoToRunWriteErrorState(); void GoToNightlyWriteErrorState(); ///Checks if a given path exist bool DoesPathExist(string path); ///Check if old run numbers can be trimmed, and if so, do it void TrimOldRunNumbers(); ///Create a given directory bool CreateDirectory(string path); /*************************************************** * INHERITED FROM DIMSERVICEINFOLIST ***************************************************/ ///Add a new service subscription void AddService(const string&, const string&, const string&, bool); ///Remove a given service subscription void RemoveService(const string&, const string&, bool); ///Remove all the services associated with a given server void RemoveAllServices(const string&); /*************************************************** * Overwritten from MessageImp ***************************************************/ vector backLogBuffer; bool shouldBackLog; public: int Write(const Time &time, const std::string &txt, int qos=kMessage); }; //DataLogger // -------------------------------------------------------------------------- // //! Overwritten write function. This way we directly log the datalogger's messages, without going through dim's dns, //! thus increasing robustness. //! @param time: see MessageImp class param //! @param txt: see MessageImp class param //! @param qos: see MessageImp class param //! @return see MessageImp class param // int DataLogger::Write(const Time&time, const std::string& txt, int qos) { if (fNightlyLogFile.is_open()) { MessageImp mimp(fNightlyLogFile); mimp.Write(time, txt, qos); } else if (shouldBackLog) { ostringstream str; MessageImp mimp(str); mimp.Write(time, txt, qos); backLogBuffer.push_back(str.str()); } return StateMachineDim::Write(time, txt, qos); } // -------------------------------------------------------------------------- // //! Check if a given path exists //! @param path the path to be checked //! @return whether or not the creation has been successfull // bool DataLogger::CreateDirectory(string path) { //remove last '/', if present if (path[path.size()-1] == '/') path = path.substr(0, path.size()-1); //create boost path const boost::filesystem::path fullPath = boost::filesystem::system_complete(boost::filesystem::path(path)); //if path does not exist, check if upper levels exist already if (boost::filesystem::exists(fullPath)) { //if path already exist, make sure it does not designate a file (filenames cannot be checked if they do not exist) if (boost::filesystem::is_directory(fullPath)) return true; Error("Path to be created contains a file name: '" + path + "'"); return false; } if (path.size() <= 1) {//we're hitting "/", which SHOULD have existed... Error("Something unexpected happened while creating a path"); } CreateDirectory(path.substr(0, path.find_last_of('/'))); //path does not exist, and upper level have been created already by recusrion. const mode_t rightsMask = S_IRWXU | S_IXGRP | S_IRGRP | S_IXOTH | S_IROTH; //everybody read, owner writes const int returnValue = mkdir(path.c_str(), rightsMask); if (returnValue != 0) { ostringstream str; str << "Could not create directory " << path << " mkdir error code: " << errno; Error(str.str()); return false; } return true; } // -------------------------------------------------------------------------- // //! Check if a given path exists //! @param path the path to be checked //! @return whether or not the given path exists // bool DataLogger::DoesPathExist(string path) { const boost::filesystem::path fullPath = boost::filesystem::system_complete(boost::filesystem::path(path)); if (!boost::filesystem::exists(fullPath)) return false; if (!boost::filesystem::is_directory(fullPath)) { Error("Path given for checking '" + path + "' designate a file name. Please provide a path name only"); return false; } if (access(path.c_str(), R_OK|W_OK|X_OK) != 0) { Error("Missing read, write or execute permissions on directory '" + path + "'"); return false; } return true; } // -------------------------------------------------------------------------- // //! Add a new service subscription //! @param server the server for which the subscription should be created //! @param service the service for which the subscription should be created //! @param isCmd whether this is a Dim Command or not. Commands are not logged // void DataLogger::AddService(const string& server, const string& service, const string&, bool isCmd) { //dataLogger does not subscribe to commands if (isCmd) return; //check the given subscription against black and white lists if (!ShouldSubscribe(server, service)) return; map &list = fServiceSubscriptions[server]; if (list.find(service) != list.end()) { Error("Service " + server + "/" + service + " is already in the dataLogger's list. ignoring its update."); return; } list[service].dimInfo.reset(SubscribeToPlease(server, service)); list[service].server = server; list[service].service = service; fNumSubAndFitsData.numSubscriptions++; if (fDebugIsOn) Debug("Added subscription to " + server + "/" + service); } // -------------------------------------------------------------------------- // //! Remove a given service subscription //! @param server the server for which the subscription should be removed //! @param service the service that should be removed //! @param isCmd whether or not this is a command // void DataLogger::RemoveService(const string& server, const string& service, bool isCmd) { if (fDestructing)//this function is called by the super class, after the destructor has deleted its own subscriptions return; if (isCmd) return; if (fServiceSubscriptions[server].erase(service) != 1) { //check the given subscription against black and white lists if (!ShouldSubscribe(server, service)) return; ostringstream str; str << "Subscription " << server << "/" << service << " could not be removed as it is not present"; Error(str.str()); return; } fNumSubAndFitsData.numSubscriptions--; if (fDebugIsOn) { Debug("Removed subscription to " + server + "/" + service); } } // -------------------------------------------------------------------------- // //! Remove all the services associated with a given server //! @param server the server for which all the services should be removed // void DataLogger::RemoveAllServices(const string& server) { fNumSubAndFitsData.numSubscriptions -= fServiceSubscriptions[server].size(); fServiceSubscriptions[server].clear(); fServiceSubscriptions.erase(server); if (fDebugIsOn) { Debug("Removed all subscriptions to " + server + "/"); } } // -------------------------------------------------------------------------- // //! Checks if the given ofstream is in error state and if so, close it //! @param out the ofstream that should be checked // bool DataLogger::CheckForOfstreamError(ofstream& out, bool isDailyStream) { if (out.good()) return true; Error("An error occured while writing to a text file. Closing it"); if (out.is_open()) out.close(); if (isDailyStream) GoToNightlyWriteErrorState(); else GoToRunWriteErrorState(); return false; } // -------------------------------------------------------------------------- // //! Open a text file and checks for error code //! @param stream the ofstream for which the file should be opened //! @name the file name // bool DataLogger::OpenTextFilePlease(ofstream& stream, const string& name) { if (stream.is_open()) return true; errno = 0; stream.open(name.c_str(), ios_base::out | ios_base::app); if (!stream.is_open()) { ostringstream str; str << "Trying to open file " << name << ": " << strerror(errno) << " (errno=" << errno << ")"; Error(str); return false; } Info("Opened: "+name); return true; } // -------------------------------------------------------------------------- // //! Create a new dim subscription to a given server and service //! @param server the server name //! @param service the service name // DimStampedInfo* DataLogger::SubscribeToPlease(const string& server, const string& service) { if (fDebugIsOn) { ostringstream str; str << "Subscribing to service " << server << "/" << service; Debug(str); } return new DimStampedInfo((server + "/" + service).c_str(), (void*)NULL, 0, this); } // -------------------------------------------------------------------------- // //! Check whether a service should be subscribed to, based on the black/white list entries //! @param server the server name associated with the service being checked //! @param service the service name associated with the service being checked // bool DataLogger::ShouldSubscribe(const string& server, const string& service) { if (fWhiteList.size()>0 && (fWhiteList.find(server + "/") == fWhiteList.end()) && (fWhiteList.find(server + "/" + service) == fWhiteList.end()) && (fWhiteList.find("/" + service) == fWhiteList.end())) return false; if ((fBlackList.find(server + "/") != fBlackList.end()) || (fBlackList.find(server + "/" + service) != fBlackList.end()) || (fBlackList.find("/" + service) != fBlackList.end())) return false; return true; } // -------------------------------------------------------------------------- // //! Compiles a file name //! @param path the base path where to put the file //! @param time the time at which the file is created //! @param service the service name, if any //! @param extension the extension to add, if any // //string DataLogger::CompileFileName(const string &path, const string &service, const string & extension, const Time &time) string DataLogger::CompileFileName(const string& service, const string& extension, const Time& time) { ostringstream str; //calculate time suitable for naming path. const Time ftime(time-boost::posix_time::time_duration(12,0,0)); //output base of file name str << Time::fmt("%Y_%m_%d") << ftime; //output service name if (!service.empty()) str << "_" << service; //output appropriate extension if (!extension.empty()) str << "." << extension; return str.str(); } string DataLogger::CompileFileNameWithPath(const string& path, const string& service, const string& extension, const Time& time) { ostringstream str; //calculate time suitable for naming files. const Time ftime(time-boost::posix_time::time_duration(12,0,0)); //output it str << path << Time::fmt("/%Y/%m/%d") << ftime; //check if target directory exist if (!DoesPathExist(str.str())) CreateDirectory(str.str()); str << '/' << CompileFileName(service, extension, time); return str.str(); } // -------------------------------------------------------------------------- // //! Compiles a file name //! @param run the run number //! @param service the service name, if any //! @param extension the extension to add, if any // string DataLogger::CompileFileName(const uint32_t run, const string& service, const string& extension) { ostringstream str; //output base of file name str << setfill('0') << setw(8) << run; //output service name if (!service.empty()) str << "_" << service; //output appropriate extension if (!extension.empty()) str << "." << extension; return str.str(); } // -------------------------------------------------------------------------- // //! Compiles a file name withh path //! @param path the base path where to put the file //! @param time the time at which the file is created //! @param run the run number //! @param service the service name, if any //! @param extension the extension to add, if any // string DataLogger::CompileFileNameWithPath(const string& path, const uint32_t run, const string& service, const string& extension, const Time& time) { ostringstream str; //calculate suitable time for naming files and output it str << path << Time::fmt("/%Y/%m/%d") << (time-boost::posix_time::time_duration(12,0,0)); //check if target directory exist if (!DoesPathExist(str.str())) CreateDirectory(str.str()); str << '/' << CompileFileName(run, service, extension);//, time); return str.str(); } // -------------------------------------------------------------------------- // //!retrieves the size on disk of a file //! @param fileName the full file name for which the size on disk should be retrieved //! @return the size of the file on disk, in bytes. 0 if the file does not exist or if an error occured // off_t DataLogger::GetFileSize(const string& fileName) { errno = 0; struct stat st; if (!stat(fileName.c_str(), &st)) return st.st_size; if (errno != 0 && errno != 2)//ignoring error #2: no such file or directory is not an error for new files { ostringstream str; str << "Unable to stat " << fileName << ". Reason: " << strerror(errno) << " [" << errno << "]"; Error(str); } return 0; } // -------------------------------------------------------------------------- // //! Removes the oldest run number and closes the fits files that should be closed //! Also creates the fits grouping file // void DataLogger::RemoveOldestRunNumber() { if (fDebugIsOn) { ostringstream str; str << "Removing run number " << fRunNumber.front().runNumber; Debug(str); } CreateFitsGrouping(fRunNumber.front().openedFits, fRunNumber.front().runNumber); //crawl through the subscriptions to see if there are still corresponding fits files opened. for (SubscriptionsListType::iterator x=fServiceSubscriptions.begin(); x!=fServiceSubscriptions.end(); x++) for (map::iterator y=x->second.begin(); y!=x->second.end(); y++) if (y->second.runFile.fRunNumber == fRunNumber.front().runNumber && y->second.runFile.IsOpen()) { y->second.runFile.Close(); } //if a grouping file is on, decrease the number of opened fits manually if (fRunNumber.front().runFitsFile) (fNumSubAndFitsData.numOpenFits)--; //remove the entry fRunNumber.pop_front(); } // -------------------------------------------------------------------------- // //! Default constructor. The name of the machine is given DATA_LOGGER //! and the state is set to kSM_Ready at the end of the function. // //!Setup the allows states, configs and transitions for the data logger // DataLogger::DataLogger(ostream &out) : StateMachineDim(out, "DATA_LOGGER"), fFilesStats("DATA_LOGGER", *this) { shouldBackLog = true; //initialize member data fNightlyFilePath = "."; fRunFilePath = "."; //Give a name to this machine's specific states AddStateName(kSM_NightlyOpen, "NightlyFileOpen", "The summary files for the night are open."); AddStateName(kSM_WaitingRun, "WaitForRun", "The summary files for the night are open and we wait for a run to be started."); AddStateName(kSM_Logging, "Logging", "The summary files for the night and the files for a single run are open."); AddStateName(kSM_BadNightlyConfig, "ErrNightlyFolder", "The folder for the nighly summary files is invalid."); AddStateName(kSM_BadRunConfig, "ErrRunFolder", "The folder for the run files is invalid."); AddStateName(kSM_DailyWriteError, "ErrDailyWrite", "An error occured while writing to a daily (and run) file."); AddStateName(kSM_RunWriteError, "ErrRunWrite", "An error occured while writing to a run file."); // Add the possible transitions for this machine AddEvent(kSM_NightlyOpen, "START", kSM_Ready, kSM_BadNightlyConfig) (bind(&DataLogger::StartPlease, this)) ("Start the nightly logging. Nightly file location must be specified already"); AddEvent(kSM_Ready, "STOP", kSM_NightlyOpen, kSM_WaitingRun, kSM_Logging, kSM_DailyWriteError, kSM_RunWriteError) (bind(&DataLogger::GoToReadyPlease, this)) ("Stop all data logging, close all files."); AddEvent(kSM_Logging, "START_RUN", kSM_WaitingRun, kSM_BadRunConfig) (bind(&DataLogger::StartRunPlease, this)) ("Start the run logging. Run file location must be specified already."); AddEvent(kSM_WaitingRun, "STOP_RUN", kSM_Logging) (bind(&DataLogger::StopRunPlease, this)) ("Wait for a run to be started, open run-files as soon as a run number arrives."); AddEvent(kSM_Ready, "RESET", kSM_Error, kSM_BadNightlyConfig, kSM_BadRunConfig, kSM_DailyWriteError, kSM_RunWriteError) (bind(&DataLogger::GoToReadyPlease, this)) ("Transition to exit error states. Closes the any open file."); AddEvent(kSM_WaitingRun, "WAIT_FOR_RUN_NUMBER", kSM_NightlyOpen, kSM_Ready) (bind(&DataLogger::NightlyToWaitRunPlease, this)) ("Go to waiting for run number state. In this state with any received run-number a new file is opened."); AddEvent(kSM_NightlyOpen, "BACK_TO_NIGHTLY_OPEN", kSM_WaitingRun) (bind(&DataLogger::BackToNightlyOpenPlease, this)) ("Go from the wait for run to nightly open state."); // Add the possible configurations for this machine AddEvent("SET_NIGHTLY_FOLDER", "C", kSM_Ready, kSM_BadNightlyConfig) (bind(&DataLogger::ConfigureNightlyFileName, this, placeholders::_1)) ("Configure the base folder for the nightly files." "|Path[string]:Absolute or relative path name where the nightly files should be stored."); AddEvent("SET_RUN_FOLDER", "C", kSM_Ready, kSM_BadNightlyConfig, kSM_NightlyOpen, kSM_WaitingRun, kSM_BadRunConfig) (bind(&DataLogger::ConfigureRunFileName, this, placeholders::_1)) ("Configure the base folder for the run files." "|Path[string]:Absolute or relative path name where the run files should be stored."); AddEvent("SET_RUN_NUMBER", "X", kSM_Ready, kSM_NightlyOpen, kSM_WaitingRun, kSM_BadRunConfig, kSM_Logging) (bind(&DataLogger::ConfigureRunNumber, this, placeholders::_1)) ("Configure the run number. Cannot be done in logging state"); // Provide a print command AddEvent("PRINT") (bind(&DataLogger::PrintStatePlease, this, placeholders::_1)) ("Print information about the internal status of the data logger."); OpenFileToDim fToDim; fToDim.code = 0; fToDim.fileName[0] = '\0'; fOpenedNightlyFiles = new DimDescribedService(GetName() + "/FILENAME_NIGHTLY", "I:1;C", fToDim, "Path and base name which is used to compile the filenames for the nightly files." "|Type[int]:type of open files (1=log, 2=rep, 4=fits)" "|Name[string]:path and base file name"); fOpenedRunFiles = new DimDescribedService(GetName() + "/FILENAME_RUN", "I:1;C", fToDim, "Path and base name which is used to compile the filenames for the run files." "|Type[int]:type of open files (1=log, 2=rep, 4=fits)" "|Name[string]:path and base file name"); fNumSubAndFitsData.numSubscriptions = 0; fNumSubAndFitsData.numOpenFits = 0; fNumSubAndFits = new DimDescribedService(GetName() + "/NUM_SUBS", "I:2", fNumSubAndFitsData, "Shows number of services to which the data logger is currently subscribed and the total number of open files." "|Subscriptions[int]:number of dim services to which the data logger is currently subscribed." "|NumOpenFiles[int]:number of files currently open by the data logger"); //services parameters fDebugIsOn = false; fOpenedFilesIsOn = true; fNumSubAndFitsIsOn = true; // provide services control commands AddEvent("SET_DEBUG_MODE", "B:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready) (bind(&DataLogger::SetDebugOnOff, this, placeholders::_1)) ("Switch debug mode on or off. Debug mode prints information about every service written to a file." "|Enable[bool]:Enable of disable debug mode (yes/no)."); AddEvent("SET_STATISTICS_UPDATE_INTERVAL", "S:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready) (bind(&DataLogger::SetStatsPeriod, this, placeholders::_1)) ("Interval in which the data-logger statistics service (STATS) is updated." "|Interval[ms]:Value in milliseconds (<=0: no update)."); AddEvent("ENABLE_FILENAME_SERVICES", "B:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready) (bind(&DataLogger::SetOpenedFilesOnOff ,this, placeholders::_1)) ("Switch service which distributes information about the open files on or off." "|Enable[bool]:Enable of disable filename services (yes/no)."); AddEvent("ENABLE_NUMSUBS_SERVICE", "B:1", kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun, kSM_Ready) (bind(&DataLogger::SetNumSubsAndFitsOnOff, this, placeholders::_1)) ("Switch the service which distributes information about the number of subscriptions and open files on or off." "|Enable[bool]:Enable of disable NUM_SUBS service (yes/no)."); AddEvent("SET_RUN_TIMEOUT", "L:1", kSM_Ready, kSM_NightlyOpen, kSM_Logging, kSM_WaitingRun) (bind(&DataLogger::SetRunTimeoutDelay, this, placeholders::_1)) ("Set the timeout delay for old run numbers." "|timeout[min]:Time out in minutes after which files for expired runs are closed."); fDestructing = false; fPreviousOldRunNumberCheck = Time().Mjd(); fDailyFileDayChangedAlready = true; fRunNumberTimeout = 60; //default run-timeout set to 1 minute if(fDebugIsOn) { Debug("DataLogger Init Done."); } } // -------------------------------------------------------------------------- // //! Destructor // DataLogger::~DataLogger() { if (fDebugIsOn) Debug("DataLogger destruction starts"); //this boolean should not be required anymore fDestructing = true; //first, let's backup the datalogger/message service subscription // shared_ptr messageBackup; // const SubscriptionsListType::iterator x = fServiceSubscriptions.find("DATA_LOGGER"); // if (x != fServiceSubscriptions.end()) // { // const map::iterator y = x->second.find("MESSAGE"); // if (y != x->second.end()) // messageBackup = y->second.dimInfo; // } // dim_lock(); //now clear the services subscriptions fServiceSubscriptions.clear(); //clear any remaining run number (should remain only one) while (fRunNumber.size() > 0) { RemoveOldestRunNumber(); } //go to the ready state. i.e. close all files, run-wise first GoToReadyPlease(); // dim_unlock(); Info("Will soon close the daily log file"); delete fOpenedNightlyFiles; delete fOpenedRunFiles; delete fNumSubAndFits; //release message service before closing nightly log file // if (messageBackup) // messageBackup.reset(); if (fNightlyLogFile.is_open())//this file is the only one that has not been closed by GoToReadyPlease { // dim_lock(); fNightlyLogFile.close(); // dim_unlock(); } if (fDebugIsOn) Debug("DataLogger desctruction ends"); } // -------------------------------------------------------------------------- // //! checks if old run numbers should be trimmed and if so, do it // void DataLogger::TrimOldRunNumbers() { const Time cTime = Time(); if (cTime - fPreviousOldRunNumberCheck < boost::posix_time::milliseconds(fRunNumberTimeout)) return; while (fRunNumber.size() > 1 && (cTime - fRunNumber.back().time) > boost::posix_time::milliseconds(fRunNumberTimeout)) { RemoveOldestRunNumber(); } fPreviousOldRunNumberCheck = cTime; } // -------------------------------------------------------------------------- // //! Inherited from DimInfo. Handles all the Infos to which we subscribed, and log them // void DataLogger::infoHandler() { DimInfo* I = getInfo(); if (I==NULL) return; //check if the service pointer corresponds to something that we subscribed to //this is a fix for a bug that provides bad Infos when a server starts bool found = false; SubscriptionsListType::iterator x; map::iterator y; for (x=fServiceSubscriptions.begin(); x != fServiceSubscriptions.end(); x++) {//find current service is subscriptions for (y=x->second.begin(); y!=x->second.end();y++) if ((y->second.dimInfo).get() == I) { found = true; break; } if (found) break; } if (!found) { DimServiceInfoList::infoHandler(); return; } if (I->getSize() <= 0 || I->getData()==NULL) { return; } if (strlen(I->getFormat()) == 0) { ostringstream str; str << "Format of " << I->getName() << " is empty. Ignoring it"; Error(str); return; } // Make sure that getTimestampMillisecs is NEVER called before // getTimestamp is properly called // check that the message has been updated by something, i.e. must be different from its initial value if (I->getTimestamp() == 0) { return; } // FIXME: Here we have to check if we have received the // service with the run-number. // CheckForRunNumber(I); has been removed because we have to // subscribe to this service anyway and hence we have the pointer // (no need to check for the name) ReportPlease(I, y->second); //remove old run numbers TrimOldRunNumbers(); } bool DataLogger::OpenStream(shared_ptr stream, const string &filename) { if (stream->is_open()) { ostringstream str; str << filename << " was already open when trying to open it."; Error(str); return false; } errno = 0; stream->open(filename.c_str(), ios_base::out | ios_base::app); if (errno != 0) { ostringstream str; str << "Unable to open " << filename << ": " << strerror(errno) << " (errno=" << errno << ")"; Error(str); return false; } if (!stream->is_open()) { ostringstream str; str << "File " << filename << " not open as it ought to be."; Error(str); return false; } Info("Opened: "+filename); return true; } // -------------------------------------------------------------------------- // //! Open the text files associated with the given run number //! @param run the run number to be dealt with // int DataLogger::OpenRunFile(RunNumberType& run) { #ifdef RUN_LOGS // open log file run.logName = CompileFileName(fRunFilePath, run.runNumber, "", "log"); if (!OpenStream(run.logFile, run.logName)) return -1; #endif // open report file run.reportName = CompileFileNameWithPath(fRunFilePath, run.runNumber, "", "rep"); if (!OpenStream(run.reportFile, run.reportName)) return -1; //get the size of the newly opened file. #ifdef RUN_LOGS fFilesStats.FileOpened(run.logName); #endif fFilesStats.FileOpened(run.reportName); //TODO this notification scheme might be messed up now.... fix it ! const string baseFileName = CompileFileNameWithPath(fRunFilePath, run.runNumber, "", ""); NotifyOpenedFile(baseFileName, 3, fOpenedRunFiles); run.openedFits.clear(); return 0; } // -------------------------------------------------------------------------- // //! Add a new active run number //! @param newRun the new run number //! @param time the time at which the new run number was issued // void DataLogger::AddNewRunNumber(int64_t newRun, Time time) { if (newRun > 0xffffffff) { Error("New run number too large, out of range. Ignoring."); return; } for (std::vector::const_iterator it=previousRunNumbers.begin(); it != previousRunNumbers.end(); it++) { if (*it == newRun) { Error("Newly provided run number has already been used (or is still in use). Going to error state"); SetCurrentState(kSM_BadRunConfig); return; } } if (fDebugIsOn) { ostringstream str; str << "Adding new run number " << newRun << " that was issued on " << time; Debug(str); } //Add new run number to run number list fRunNumber.push_back(RunNumberType()); fRunNumber.back().runNumber = uint32_t(newRun); fRunNumber.back().time = time; ostringstream str; str << "The new run number is: " << fRunNumber.back().runNumber; Message(str); if (GetCurrentState() != kSM_Logging) return; //open the log and report files if (OpenRunFile(fRunNumber.back()) != 0) {//an error occured. close current run files and go to error state for (list::iterator it=fRunNumber.begin(); it != fRunNumber.end(); it++) { if (it->reportFile->is_open()) { it->reportFile->close(); Info("Closed: "+it->reportName); } #ifdef RUN_LOGS if (it->logFile->is_open()) { it->logFile->close(); Info("Closed: "+it->logName); } #endif } StopRunPlease(); SetCurrentState(kSM_BadRunConfig); } } // -------------------------------------------------------------------------- // //! Checks whether or not the current info is a run number. //! If so, then remember it. A run number is required to open the run-log file //! @param I //! the current DimInfo // void DataLogger::CheckForRunNumber(DimInfo* I) { if (strstr(I->getName(), "SET_RUN_NUMBER") != NULL) {//assumes that the run number is an integer //check if some run number entries can be deleted leave one so that two remain after adding the new one AddNewRunNumber(I->getLonglong(), Time(I->getTimestamp(), I->getTimestampMillisecs()*1000)); } } // -------------------------------------------------------------------------- // //! write infos to log files. //! @param I //! The current DimInfo //! @param sub //! The dataLogger's subscription corresponding to this DimInfo // void DataLogger::ReportPlease(DimInfo* I, SubscriptionType& sub) { const string fmt(I->getFormat()); const bool isItaReport = fmt!="C"; if (!fNightlyLogFile.is_open()) return; if (fDebugIsOn && string(I->getName())!="DATA_LOGGER/MESSAGE") { ostringstream str; str << "Logging " << I->getName() << " [" << I->getFormat() << "] (" << I->getSize() << ")"; Debug(str); } //Check whether we should close and reopen daily text files or not //This should work in any case base of the following: // - fDailyFileDayChangedAlready is initialized to true. So if the dataLogger is started around noon, no file will be closed // - fDailyFileDayChangedAlready is set to false if (time != 12), so the file will be closed and reopened only if the logger runs since before noon (which is the required behavior) //This only applies to text files. Fits are closed and reopened based on the last and current service received time. //this was not applicable to text files, because as they gather several services, we have no guarantee that the received time will be greater than the previous one, //which could lead to several close/reopen instead of only one. if (Time().h() == 12 && !fDailyFileDayChangedAlready) { if (fDebugIsOn) Debug("Its Noon! Closing and reopening nightly text files"); fNightlyLogFile.close(); fNightlyReportFile.close(); Info("Closed: "+fFullNightlyLogFileName); Info("Closed: "+fFullNightlyReportFileName); fFullNightlyLogFileName = CompileFileNameWithPath(fNightlyFilePath, "", "log"); if (!OpenTextFilePlease(fNightlyLogFile, fFullNightlyLogFileName)) { GoToReadyPlease(); SetCurrentState(kSM_BadNightlyConfig); return; } fFullNightlyReportFileName = CompileFileNameWithPath(fNightlyFilePath, "", "rep"); if (!OpenTextFilePlease(fNightlyReportFile, fFullNightlyReportFileName)) { GoToReadyPlease(); SetCurrentState(kSM_BadNightlyConfig); return; } fDailyFileDayChangedAlready = true; } if (Time().h() != 12 && fDailyFileDayChangedAlready) fDailyFileDayChangedAlready = false; //create the converter for that service if (!sub.fConv.get()) { sub.fConv = shared_ptr(new Converter(Out(), I->getFormat())); if (!sub.fConv) { ostringstream str; str << "Couldn't properly parse the format... service " << sub.dimInfo->getName() << " ignored."; Error(str); return; } } //construct the header ostringstream header; const Time cTime(I->getTimestamp(), I->getTimestampMillisecs()*1000); fQuality = I->getQuality(); fMjD = cTime.Mjd(); //figure out which run file should be used ofstream* targetRunFile = NULL; RunNumberType* cRunNumber = NULL; if (GetCurrentState() == kSM_Logging) { list::reverse_iterator rit; for (rit=fRunNumber.rbegin(); rit!=fRunNumber.rend(); rit++) { if (rit->time < cTime) //this is the run number that we want to use { //Find something better to convert iterator to pointer than the ugly line below.... cRunNumber = &(*rit); sub.runNumber = rit->runNumber; #ifdef RUN_LOGS targetRunFile = isItaReport ? (rit->reportFile).get() : (rit->logFile).get(); #else targetRunFile = isItaReport ? (rit->reportFile).get() : NULL; #endif break; } } if (rit == fRunNumber.rend() && fRunNumber.size() != 0) { ostringstream str; str << "Could not find an appropriate run number for info coming at time: " << cTime; Error(str); Error("Active run numbers: "); for (rit=fRunNumber.rbegin(); rit != fRunNumber.rend(); rit++) { str.str(""); str << rit->runNumber; Error(str); } } } if (isItaReport) { //write text header header << I->getName() << " " << fQuality << " "; header << cTime.Y() << " " << cTime.M() << " " << cTime.D() << " "; header << cTime.h() << " " << cTime.m() << " " << cTime.s() << " "; header << cTime.ms() << " " << I->getTimestamp() << " "; string text; try { text = sub.fConv->GetString(I->getData(), I->getSize()); } catch (const runtime_error &e) { ostringstream str; str << "Parsing service " << sub.dimInfo->getName(); str << " failed: " << e.what() << " removing the subscription for now."; Error(str); //remove this subscription from the list. //because these operators use references to elements, and because they're supposed here to erase these objects on the way, I'm not too sure... so duplicate the names ! string server = sub.server; string service = sub.service; fServiceSubscriptions.find(server)->second.erase(service); return; } if (text.empty()) { ostringstream str; str << "Service " << sub.dimInfo->getName() << " sent an empty string"; Info(str); return; } //replace bizarre characters by white space replace(text.begin(), text.end(), '\n', '\\'); replace_if(text.begin(), text.end(), ptr_fun(&iscntrl), ' '); //write entry to Nightly report if (fNightlyReportFile.is_open()) { fNightlyReportFile << header.str() << text << endl; if (!CheckForOfstreamError(fNightlyReportFile, true)) return; } //write entry to run-report if (targetRunFile && targetRunFile->is_open()) { *targetRunFile << header.str() << text << endl; if (!CheckForOfstreamError(*targetRunFile, false)) return; } #ifdef HAVE_FITS //check if the last received event was before noon and if current one is after noon. //if so, close the file so that it gets reopened. if (sub.nightlyFile.IsOpen()) if ((sub.lastReceivedEvent != Time::None) && (sub.lastReceivedEvent.h() < 12) && (cTime.h() >= 12)) { sub.nightlyFile.Close(); } sub.lastReceivedEvent = cTime; if (!sub.nightlyFile.IsOpen() || !sub.runFile.IsOpen() || sub.runNumber != sub.runFile.fRunNumber) if (GetCurrentState() != kSM_Ready) OpenFITSFilesPlease(sub, cRunNumber); WriteToFITS(sub); #endif } else {//write entry to both Nightly and run logs vector strings; try { strings = sub.fConv->ToStrings(I->getData()); } catch (const runtime_error &e) { ostringstream str; str << "Parsing service " << sub.dimInfo->getName(); str << " failed: " << e.what() << " removing the subscription for now."; Error(str); //remove this subscription from the list. //because these operators use references to elements, and because they're supposed here to erase these objects on the way, I'm not too sure... so duplicate the names ! string server = sub.server; string service = sub.service; fServiceSubscriptions.find(server)->second.erase(service); return; } if (strings.size() > 1) { ostringstream err; err << "There was more than one string message in service " << I->getName() << " going to fatal error state"; Error(err.str()); } ostringstream msg; msg << I->getName() << ": " << strings[0]; if (fNightlyLogFile.is_open()) { MessageImp(fNightlyLogFile).Write(cTime, msg.str().c_str(), fQuality); if (!CheckForOfstreamError(fNightlyLogFile, true)) return; } if (targetRunFile && targetRunFile->is_open()) { MessageImp(*targetRunFile).Write(cTime, msg.str().c_str(), fQuality); if (!CheckForOfstreamError(*targetRunFile, false)) return; } } } // -------------------------------------------------------------------------- // //! print the dataLogger's current state. invoked by the PRINT command //! @param evt //! the current event. Not used by the method //! @returns //! the new state. Which, in that case, is the current state //! int DataLogger::PrintStatePlease(const Event& ) { Message("------------------------------------------"); Message("------- DATA LOGGER CURRENT STATE --------"); Message("------------------------------------------"); //print the path configuration Message("Nightly path: " + boost::filesystem::system_complete(boost::filesystem::path(fNightlyFilePath)).directory_string()); Message("Run path: " + boost::filesystem::system_complete(boost::filesystem::path(fRunFilePath)).directory_string()); //print active run numbers ostringstream str; //timeout value str << "Timeout delay for old run numbers: " << fRunNumberTimeout << " ms"; Message(str); str.str(""); str << "Active Run Numbers:"; for (list::const_iterator it=fRunNumber.begin(); it!=fRunNumber.end(); it++) str << " " << it->runNumber; if (fRunNumber.size()==0) str << " "; Message(str); //print all the open files. Message("------------ OPEN FILES ----------------"); if (fNightlyLogFile.is_open()) Message("Nightly log-file: "+fFullNightlyLogFileName); if (fNightlyReportFile.is_open()) Message("Nightly report-file: "+fFullNightlyReportFileName); for (list::const_iterator it=fRunNumber.begin(); it!=fRunNumber.end(); it++) { #ifdef RUN_LOGS if (it->logFile->is_open()) Message("Run log-file: " + it->logName); #endif if (it->reportFile->is_open()) Message("Run report-file: " + it->reportName); } const DimWriteStatistics::Stats statVar = fFilesStats.GetTotalSizeWritten(); // /*const bool statWarning =*/ calculateTotalSizeWritten(statVar, true); #ifdef HAVE_FITS str.str(""); str << "Number of open FITS files: " << fNumSubAndFitsData.numOpenFits; Message(str); // FIXME: Print list of open FITS files #else Message("FITS output disabled at compilation"); #endif Message("----------------- STATS ------------------"); if (fFilesStats.GetUpdateInterval()>0) { str.str(""); str << "Statistics are updated every " << fFilesStats.GetUpdateInterval() << " ms"; Message(str); } else Message("Statistics updates are currently disabled."); str.str(""); str << "Total Size written: " << statVar.sizeWritten/1000 << " kB"; Message(str); str.str(""); str << "Disk free space: " << statVar.freeSpace/1000000 << " MB"; Message(str); Message("------------ DIM SUBSCRIPTIONS -----------"); str.str(""); str << "There are " << fNumSubAndFitsData.numSubscriptions << " active DIM subscriptions."; Message(str); for (map >::const_iterator it=fServiceSubscriptions.begin(); it!= fServiceSubscriptions.end();it++) { Message("Server "+it->first); for (map::const_iterator it2=it->second.begin(); it2!=it->second.end(); it2++) Message(" -> "+it2->first); } Message("--------------- BLOCK LIST ---------------"); for (set::const_iterator it=fBlackList.begin(); it != fBlackList.end(); it++) Message(" -> "+*it); if (fBlackList.size()==0) Message(" "); Message("--------------- ALLOW LIST ---------------"); for (set::const_iterator it=fWhiteList.begin(); it != fWhiteList.end(); it++) Message(" -> "+*it); if (fWhiteList.size()==0) Message(" "); Message("-------------- GROUPING LIST -------------"); Message("The following servers and/or services will"); Message("be grouped into a single fits file:"); for (set::const_iterator it=fGrouping.begin(); it != fGrouping.end(); it++) Message(" -> "+*it); if (fGrouping.size()==0) Message(" "); Message("------------------------------------------"); Message("-------- END OF DATA LOGGER STATE --------"); Message("------------------------------------------"); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! turn debug mode on and off //! @param evt //! the current event. contains the instruction string: On, Off, on, off, ON, OFF, 0 or 1 //! @returns //! the new state. Which, in that case, is the current state //! int DataLogger::SetDebugOnOff(const Event& evt) { const bool backupDebug = fDebugIsOn; fDebugIsOn = evt.GetBool(); if (fDebugIsOn == backupDebug) Message("Debug mode was already in the requested state."); ostringstream str; str << "Debug mode is now " << fDebugIsOn; Message(str); fFilesStats.SetDebugMode(fDebugIsOn); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! set the statistics update period duration. 0 disables the statistics //! @param evt //! the current event. contains the new duration. //! @returns //! the new state. Which, in that case, is the current state //! int DataLogger::SetStatsPeriod(const Event& evt) { fFilesStats.SetUpdateInterval(evt.GetShort()); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! set the opened files service on or off. //! @param evt //! the current event. contains the instruction string. similar to setdebugonoff //! @returns //! the new state. Which, in that case, is the current state //! int DataLogger::SetOpenedFilesOnOff(const Event& evt) { const bool backupOpened = fOpenedFilesIsOn; fOpenedFilesIsOn = evt.GetBool(); if (fOpenedFilesIsOn == backupOpened) Message("Opened files service mode was already in the requested state."); ostringstream str; str << "Opened files service mode is now " << fOpenedFilesIsOn; Message(str); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! set the number of subscriptions and opened fits on and off //! @param evt //! the current event. contains the instruction string. similar to setdebugonoff //! @returns //! the new state. Which, in that case, is the current state //! int DataLogger::SetNumSubsAndFitsOnOff(const Event& evt) { const bool backupSubs = fNumSubAndFitsIsOn; fNumSubAndFitsIsOn = evt.GetBool(); if (fNumSubAndFitsIsOn == backupSubs) Message("Number of subscriptions service mode was already in the requested state"); ostringstream str; str << "Number of subscriptions service mode is now " << fNumSubAndFitsIsOn; Message(str); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! set the timeout delay for old run numbers //! @param evt //! the current event. contains the timeout delay long value //! @returns //! the new state. Which, in that case, is the current state //! int DataLogger::SetRunTimeoutDelay(const Event& evt) { if (evt.GetUInt() == 0) { Error("Timeout delays for old run numbers must be greater than 0... ignored."); return GetCurrentState(); } if (fRunNumberTimeout == evt.GetUInt()) Message("New timeout for old run numbers is same value as previous one."); fRunNumberTimeout = evt.GetUInt(); ostringstream str; str << "Timeout delay for old run numbers is now " << fRunNumberTimeout << " ms"; Message(str); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! Configure a given file name //! @param target //! where to put the result //! @param type //! what to append to the created path. currently only run or nightly //! @param evt //! the event transporting the path //! @returns //! currently only the current state. // int DataLogger::ConfigureFileName(string &target, const string &type, const EventImp &evt) { if (!evt.GetText()) { Error("Empty "+type+" folder given. Please specify a valid path."); return GetCurrentState(); } const string givenPath = evt.GetText(); if (!DoesPathExist(givenPath)) { Error("Provided "+type+" path '"+givenPath+"' is not a valid folder. Ignored"); return GetCurrentState(); } Message("New "+type+" folder: "+givenPath); target = givenPath; fFilesStats.SetCurrentFolder(givenPath); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! Sets the path to use for the Nightly log file. //! @param evt //! the event transporting the path //! @returns //! currently only the current state. // int DataLogger::ConfigureNightlyFileName(const Event& evt) { return ConfigureFileName(fNightlyFilePath, "nightly", evt); } // -------------------------------------------------------------------------- // //! Sets the path to use for the run log file. //! @param evt //! the event transporting the path //! @returns //! currently only the current state int DataLogger::ConfigureRunFileName(const Event& evt) { return ConfigureFileName(fRunFilePath, "run", evt); } // -------------------------------------------------------------------------- // //! Sets the run number. //! @param evt //! the event transporting the run number //! @returns //! currently only the current state int DataLogger::ConfigureRunNumber(const Event& evt) { AddNewRunNumber(evt.GetXtra(), evt.GetTime()); return GetCurrentState(); } // -------------------------------------------------------------------------- // //! Notifies the DIM service that a particular file was opened //! @ param name the base name of the opened file, i.e. without path nor extension. //! WARNING: use string instead of string& because I pass values that do not convert to string&. //! this is not a problem though because file are not opened so often. //! @ param type the type of the opened file. 0 = none open, 1 = log, 2 = text, 4 = fits inline void DataLogger::NotifyOpenedFile(const string &name, int type, DimDescribedService* service) { if (!fOpenedFilesIsOn) return; if (fDebugIsOn) { ostringstream str; str << "Updating " << service->getName() << " file '" << name << "' (type=" << type << ")"; Debug(str); str.str(""); str << "Num subscriptions: " << fNumSubAndFitsData.numSubscriptions << " Num open FITS files: " << fNumSubAndFitsData.numOpenFits; Debug(str); } if (name.size()+1 > FILENAME_MAX) { Error("Provided file name '" + name + "' is longer than allowed file name length."); return; } OpenFileToDim fToDim; fToDim.code = type; memcpy(fToDim.fileName, name.c_str(), name.size()+1); service->setData(reinterpret_cast(&fToDim), name.size()+1+sizeof(int)); service->setQuality(0); service->updateService(); } // -------------------------------------------------------------------------- // //! Implements the Start transition. //! Concatenates the given path for the Nightly file and the filename itself (based on the day), //! and tries to open it. //! @returns //! kSM_NightlyOpen if success, kSM_BadNightlyConfig if failure int DataLogger::StartPlease() { if (fDebugIsOn) { Debug("Starting..."); } fFullNightlyLogFileName = CompileFileNameWithPath(fNightlyFilePath, "", "log"); if (!OpenTextFilePlease(fNightlyLogFile, fFullNightlyLogFileName)) return kSM_BadNightlyConfig; fFullNightlyReportFileName = CompileFileNameWithPath(fNightlyFilePath, "", "rep"); if (!OpenTextFilePlease(fNightlyReportFile, fFullNightlyReportFileName)) { fNightlyLogFile.close(); Info("Closed: "+fFullNightlyReportFileName); return kSM_BadNightlyConfig; } fFilesStats.FileOpened(fFullNightlyLogFileName); fFilesStats.FileOpened(fFullNightlyReportFileName); //notify that a new file has been opened. const string baseFileName = CompileFileNameWithPath(fNightlyFilePath, "", ""); NotifyOpenedFile(baseFileName, 3, fOpenedNightlyFiles); fOpenedNightlyFits.clear(); return kSM_NightlyOpen; } #ifdef HAVE_FITS // -------------------------------------------------------------------------- // //! open if required a the FITS files corresponding to a given subscription //! @param sub //! the current DimInfo subscription being examined void DataLogger::OpenFITSFilesPlease(SubscriptionType& sub, RunNumberType* cRunNumber) { string serviceName(sub.dimInfo->getName()); //if run number has changed, reopen a new fits file with the correct run number. if (sub.runFile.IsOpen() && sub.runFile.fRunNumber != sub.runNumber) { sub.runFile.Close(); Info("Closed: "+sub.runFile.fFileName+" (new run number)"); } //we must check if we should group this service subscription to a single fits file before we replace the / by _ bool hasGrouping = false; if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging)) {//will we find this service in the grouping list ? for (set::const_iterator it=fGrouping.begin(); it!=fGrouping.end(); it++) { if (serviceName.find(*it) != string::npos) { hasGrouping = true; break; } } } for (unsigned int i=0;iupdateService(); } //do the actual file open if (!sub.runFile.IsOpen() && (GetCurrentState() == kSM_Logging) && sub.runNumber != 0) {//buffer for the run file have already been allocated when doing the Nightly file string fileNameOnly; string partialName; if (hasGrouping) { partialName = CompileFileNameWithPath(fRunFilePath, sub.runNumber, "", "fits"); fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size()); } else { partialName = CompileFileNameWithPath(fRunFilePath, sub.runNumber, serviceName, "fits"); fileNameOnly = partialName.substr(partialName.find_last_of('/')+1, partialName.size()); } //get the size of the file we're about to open if (fFilesStats.FileOpened(partialName)) cRunNumber->openedFits[fileNameOnly].push_back(serviceName); else if (hasGrouping) { cRunNumber->addServiceToOpenedFits(fileNameOnly, serviceName); } if (hasGrouping && (!cRunNumber->runFitsFile.get())) try { cRunNumber->runFitsFile = shared_ptr(new CCfits::FITS(partialName, CCfits::RWmode::Write)); (fNumSubAndFitsData.numOpenFits)++; } catch (CCfits::FitsException e) { ostringstream str; str << "Open FITS file " << partialName << ": " << e.message(); Error(str); cRunNumber->runFitsFile = shared_ptr(); GoToRunWriteErrorState(); return; } const string baseFileName = CompileFileNameWithPath(fRunFilePath, sub.runNumber, "", ""); NotifyOpenedFile(baseFileName, 7, fOpenedRunFiles); if (hasGrouping) { if (!sub.runFile.Open(partialName, serviceName, (cRunNumber->runFitsFile).get(), &fNumSubAndFitsData.numOpenFits, this, sub.runNumber)) { GoToRunWriteErrorState(); return; } } else { if (!sub.runFile.Open(partialName, serviceName, NULL, &fNumSubAndFitsData.numOpenFits, this, sub.runNumber)) { GoToRunWriteErrorState(); return; } } ostringstream str; str << "Opened: " << partialName << " (Nfits=" << fNumSubAndFitsData.numOpenFits << ")"; Info(str); if (fNumSubAndFitsIsOn) fNumSubAndFits->updateService(); } } // -------------------------------------------------------------------------- // //! Allocates the required memory for a given pair of fits files (nightly and run) //! @param sub the subscription of interest. // void DataLogger::AllocateFITSBuffers(SubscriptionType& sub) { //Init the time columns of the file Description dateDesc(string("Time"), string("Modified Julian Date"), string("MJD")); sub.nightlyFile.AddStandardColumn(dateDesc, "1D", &fMjD, sizeof(double)); sub.runFile.AddStandardColumn(dateDesc, "1D", &fMjD, sizeof(double)); Description QoSDesc("QoS", "Quality of service", ""); sub.nightlyFile.AddStandardColumn(QoSDesc, "1J", &fQuality, sizeof(int)); sub.runFile.AddStandardColumn(QoSDesc, "1J", &fQuality, sizeof(int)); const Converter::FormatList flist = sub.fConv->GetList(); // Compilation failed if (!sub.fConv->valid()) { Error("Compilation of format string failed."); return; } //we've got a nice structure describing the format of this service's messages. //Let's create the appropriate FITS columns int size = 0; vector dataFormatsLocal; for (unsigned int i=0;iname()[0]) { case 'c': dataQualifier << 'B'; break; case 's': dataQualifier << 'I'; break; case 'i': dataQualifier << 'J'; break; case 'l': dataQualifier << 'J'; break; case 'f': dataQualifier << 'E'; break; case 'd': dataQualifier << 'D'; break; case 'x': dataQualifier << 'K'; break; case 'S': //we skip the variable length strings continue; default: Fatal("THIS SHOULD NEVER BE REACHED."); }; size += flist[i].first.second * flist[i].second.first; dataFormatsLocal.push_back(dataQualifier.str()); } sub.nightlyFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), size); sub.runFile.InitDataColumns(GetDescription(sub.server, sub.service), dataFormatsLocal, sub.dimInfo->getData(), size); sub.fitsBufferAllocated = true; } // -------------------------------------------------------------------------- // //! write a dimInfo data to its corresponding FITS files // void DataLogger::WriteToFITS(SubscriptionType& sub) { //nightly File status (open or not) already checked if (sub.nightlyFile.IsOpen()) { if (!sub.nightlyFile.Write(sub.fConv.get())) { sub.nightlyFile.Close(); GoToNightlyWriteErrorState(); return; } } if (sub.runFile.IsOpen()) { if (!sub.runFile.Write(sub.fConv.get())) { sub.runFile.Close(); GoToRunWriteErrorState(); return; } } } #endif //if has_fits // -------------------------------------------------------------------------- // //! Go to Run Write Error State // A write error has occurred. Checks what is the current state and take appropriate action void DataLogger::GoToRunWriteErrorState() { if ((GetCurrentState() != kSM_RunWriteError) && (GetCurrentState() != kSM_DailyWriteError)) SetCurrentState(kSM_RunWriteError); } // -------------------------------------------------------------------------- // //! Go to Nightly Write Error State // A write error has occurred. Checks what is the current state and take appropriate action void DataLogger::GoToNightlyWriteErrorState() { if (GetCurrentState() != kSM_DailyWriteError) SetCurrentState(kSM_DailyWriteError); } // -------------------------------------------------------------------------- // //! Implements the StartRun transition. //! Concatenates the given path for the run file and the filename itself (based on the run number), //! and tries to open it. //! @returns //! kSM_Logging if success, kSM_BadRunConfig if failure. int DataLogger::StartRunPlease() { if (fDebugIsOn) { Debug("Starting Run Logging..."); } //open all the relevant run-files. i.e. all the files associated with run numbers. for (list::iterator it=fRunNumber.begin(); it != fRunNumber.end(); it++) if (OpenRunFile(*it) != 0) { StopRunPlease(); return kSM_BadRunConfig; } return kSM_Logging; } #ifdef HAVE_FITS // -------------------------------------------------------------------------- // //! Create a fits group file with all the run-fits that were written (either daily or run) //! @param filesToGroup a map of filenames mapping to table names to be grouped (i.e. a //! single file can contain several tables to group //! @param runNumber the run number that should be used for grouping. 0 means nightly group // void DataLogger::CreateFitsGrouping(map > & filesToGroup, int runNumber) { if (fDebugIsOn) { ostringstream str; str << "Creating fits group for "; if (runNumber != 0) str << "run files"; else str << "nightly files"; Debug(str); } //create the FITS group corresponding to the ending run. CCfits::FITS* groupFile; unsigned int numFilesToGroup = 0; unsigned int maxCharLength = 0; for (map >::const_iterator it=filesToGroup.begin(); it != filesToGroup.end(); it++) { //add the number of tables in this file to the total number to group numFilesToGroup += it->second.size(); //check the length of all the strings to be written, to determine the max string length to write if (it->first.size() > maxCharLength) maxCharLength = it->first.size(); for (vector::const_iterator jt=it->second.begin(); jt != it->second.end(); jt++) if (jt->size() > maxCharLength) maxCharLength = jt->size(); } if (fDebugIsOn) { ostringstream str; str << "There are " << numFilesToGroup << " tables to group"; Debug(str); } if (numFilesToGroup <= 1) { filesToGroup.clear(); return; } string groupName; if (runNumber != 0) groupName = CompileFileNameWithPath(fRunFilePath, runNumber, "", "fits"); else groupName = CompileFileNameWithPath(fNightlyFilePath, "", "fits"); Info("Creating FITS group in: "+groupName); CCfits::Table* groupTable; // const int maxCharLength = FILENAME_MAX; try { groupFile = new CCfits::FITS(groupName, CCfits::RWmode::Write); //setup the column names ostringstream pathTypeName; pathTypeName << maxCharLength << "A"; vector names; vector dataTypes; names.push_back("MEMBER_XTENSION"); dataTypes.push_back("8A"); names.push_back("MEMBER_URI_TYPE"); dataTypes.push_back("3A"); names.push_back("MEMBER_LOCATION"); dataTypes.push_back(pathTypeName.str()); names.push_back("MEMBER_NAME"); dataTypes.push_back(pathTypeName.str()); groupTable = groupFile->addTable("GROUPING", numFilesToGroup, names, dataTypes); //TODO handle the case when the logger was stopped and restarted during the same day, i.e. the grouping file must be updated } catch (CCfits::FitsException e) { ostringstream str; str << "Creating FITS table GROUPING in " << groupName << ": " << e.message(); Error(str); return; } //CCfits seems to be buggy somehow: can't use the column's function "write": it create a compilation error: maybe strings were not thought about. //use cfitsio routines instead groupTable->makeThisCurrent(); //create appropriate buffer. const unsigned int n = 8 + 3 + 2*maxCharLength + 1; //+1 for trailling character vector realBuffer; realBuffer.resize(n); unsigned char* fitsBuffer = &realBuffer[0]; memset(fitsBuffer, 0, n); char* startOfExtension = reinterpret_cast(fitsBuffer); char* startOfURI = reinterpret_cast(&fitsBuffer[8]); char* startOfLocation = reinterpret_cast(&fitsBuffer[8 + 3]); char* startOfName = reinterpret_cast(&fitsBuffer[8+3+maxCharLength]); strcpy(startOfExtension, "BINTABLE"); strcpy(startOfURI, "URL"); int i=1; for (map >::const_iterator it=filesToGroup.begin(); it!=filesToGroup.end(); it++) for (vector::const_iterator jt=it->second.begin(); jt != it->second.end(); jt++, i++) { strcpy(startOfLocation, it->first.c_str()); strcpy(startOfName, jt->c_str()); if (fDebugIsOn) { ostringstream str; str << "Grouping " << it->first << " " << *jt; Debug(str); } int status = 0; fits_write_tblbytes(groupFile->fitsPointer(), i, 1, 8+3+2*maxCharLength, fitsBuffer, &status); if (status) { char text[30];//max length of cfitsio error strings (from doc) fits_get_errstatus(status, text); ostringstream str; str << "Writing FITS row " << i << " in " << groupName << ": " << text << " (file_write_tblbytes, rc=" << status << ")"; Error(str); GoToRunWriteErrorState(); delete groupFile; return; } } filesToGroup.clear(); delete groupFile; } #endif //HAVE_FITS // -------------------------------------------------------------------------- // //! Implements the StopRun transition. //! Attempts to close the run file. //! @returns //! kSM_WaitingRun if success, kSM_FatalError otherwise int DataLogger::StopRunPlease() { if (fDebugIsOn) { Debug("Stopping Run Logging..."); } //it may be that dim tries to write a dimInfo at the same time as we're closing files. Prevent this // dim_lock(); for (list::const_iterator it=fRunNumber.begin(); it != fRunNumber.end(); it++) { #ifdef RUN_LOGS if (!it->logFile->is_open() || !it->reportFile->is_open()) #else if (!it->reportFile->is_open()) #endif return kSM_FatalError; #ifdef RUN_LOGS it->logFile->close(); Info("Closed: "+it->logName); #endif it->reportFile->close(); Info("Closed: "+it->reportName); } #ifdef HAVE_FITS for (SubscriptionsListType::iterator i = fServiceSubscriptions.begin(); i != fServiceSubscriptions.end(); i++) for (map::iterator j = i->second.begin(); j != i->second.end(); j++) { if (j->second.runFile.IsOpen()) j->second.runFile.Close(); } #endif NotifyOpenedFile("", 0, fOpenedRunFiles); if (fNumSubAndFitsIsOn) fNumSubAndFits->updateService(); while (fRunNumber.size() > 0) { RemoveOldestRunNumber(); } // dim_unlock(); return kSM_WaitingRun; } // -------------------------------------------------------------------------- // //! Implements the Stop and Reset transitions. //! Attempts to close any openned file. //! @returns //! kSM_Ready int DataLogger::GoToReadyPlease() { if (fDebugIsOn) { Debug("Going to the Ready state..."); } if (GetCurrentState() == kSM_Logging) StopRunPlease(); //it may be that dim tries to write a dimInfo while we're closing files. Prevent that const string baseFileName = CompileFileNameWithPath(fNightlyFilePath, "", ""); if (fNightlyReportFile.is_open()) { fNightlyReportFile.close(); Info("Closed: "+baseFileName+".rep"); } #ifdef HAVE_FITS for (SubscriptionsListType::iterator i = fServiceSubscriptions.begin(); i != fServiceSubscriptions.end(); i++) for (map::iterator j = i->second.begin(); j != i->second.end(); j++) { if (j->second.nightlyFile.IsOpen()) j->second.nightlyFile.Close(); } #endif if (GetCurrentState() == kSM_Logging || GetCurrentState() == kSM_WaitingRun || GetCurrentState() == kSM_NightlyOpen) { NotifyOpenedFile("", 0, fOpenedNightlyFiles); if (fNumSubAndFitsIsOn) fNumSubAndFits->updateService(); } #ifdef HAVE_FITS CreateFitsGrouping(fOpenedNightlyFits, 0); #endif return kSM_Ready; } // -------------------------------------------------------------------------- // //! Implements the transition towards kSM_WaitingRun //! If current state is kSM_Ready, then tries to go to nightlyOpen state first. //! @returns //! kSM_WaitingRun or kSM_badNightlyConfig int DataLogger::NightlyToWaitRunPlease() { int cState = GetCurrentState(); if (cState == kSM_Ready) cState = StartPlease(); if (cState != kSM_NightlyOpen) return GetCurrentState(); if (fDebugIsOn) { Debug("Going to Wait Run Number state..."); } return kSM_WaitingRun; } // -------------------------------------------------------------------------- // //! Implements the transition from wait for run number to nightly open //! Does nothing really. //! @returns //! kSM_WaitingRun int DataLogger::BackToNightlyOpenPlease() { if (fDebugIsOn) { Debug("Going to NightlyOpen state..."); } return kSM_NightlyOpen; } // -------------------------------------------------------------------------- // //! Setup Logger's configuration from a Configuration object //! @param conf the configuration object that should be used //! int DataLogger::EvalConfiguration(const Configuration& conf) { fDebugIsOn = conf.Get("debug"); fFilesStats.SetDebugMode(fDebugIsOn); //Set the block or allow list fBlackList.clear(); fWhiteList.clear(); //Adding entries that should ALWAYS be ignored fBlackList.insert("DATA_LOGGER/MESSAGE"); fBlackList.insert("/SERVICE_LIST"); fBlackList.insert("DIS_DNS/"); //set the black list, white list and the goruping const vector vec1 = conf.Vec("block"); const vector vec2 = conf.Vec("allow"); const vector vec3 = conf.Vec("group"); fBlackList.insert(vec1.begin(), vec1.end()); fWhiteList.insert(vec2.begin(), vec2.end()); fGrouping.insert( vec3.begin(), vec3.end()); //set the old run numbers timeout delay if (conf.Has("run-timeout")) { const uint32_t timeout = conf.Get("run-timeout"); if (timeout == 0) { Error("Time out delay for old run numbers must not be 0."); return 1; } fRunNumberTimeout = timeout; } //configure the run files directory if (conf.Has("destination-folder")) { const string folder = conf.Get("destination-folder"); if (!fFilesStats.SetCurrentFolder(folder)) return 2; fRunFilePath = folder; fNightlyFilePath = folder; fFullNightlyLogFileName = CompileFileNameWithPath(fNightlyFilePath, "", "log"); if (!OpenTextFilePlease(fNightlyLogFile, fFullNightlyLogFileName)) SetCurrentState(kSM_BadNightlyConfig); else { NotifyOpenedFile(fFullNightlyLogFileName, 1, fOpenedNightlyFiles); for (vector::iterator it=backLogBuffer.begin();it!=backLogBuffer.end();it++) { fNightlyLogFile << *it; } } } shouldBackLog = false; backLogBuffer.clear(); //configure the interval between statistics updates if (conf.Has("stats-interval")) fFilesStats.SetUpdateInterval(conf.Get("stats-interval")); //configure if the filenames service is on or off fOpenedFilesIsOn = !conf.Get("no-filename-service"); //configure if the number of subscriptions and fits files is on or off. fNumSubAndFitsIsOn = !conf.Get("no-numsubs-service"); return -1; } #include "Main.h" // -------------------------------------------------------------------------- template int RunShell(Configuration &conf) { return Main(conf, true); } /* Extract usage clause(s) [if any] for SYNOPSIS. Translators: "Usage" and "or" here are patterns (regular expressions) which are used to match the usage synopsis in program output. An example from cp (GNU coreutils) which contains both strings: Usage: cp [OPTION]... [-T] SOURCE DEST or: cp [OPTION]... SOURCE... DIRECTORY or: cp [OPTION]... -t DIRECTORY SOURCE... */ void PrintUsage() { cout << "\n" "The data logger connects to all available Dim services and " "writes them to ascii and fits files.\n" "\n" "The default is that the program is started without user interaction. " "All actions are supposed to arrive as DimCommands. Using the -c " "option, a local shell can be initialized. With h or help a short " "help message about the usage can be brought to the screen.\n" "\n" "Usage: datalogger [-c type] [OPTIONS]\n" " or: datalogger [OPTIONS]\n"; cout << endl; } // -------------------------------------------------------------------------- void PrintHelp() { /* Additional help text which is printed after the configuration options goes here */ cout << "\n" "If the allow list has any element, only the servers and/or services " "specified in the list will be used for subscription. The black list " "will disable service subscription and has higher priority than the " "allow list. If the allow list is not present by default all services " "will be subscribed." "\n" "For example, block=DIS_DNS/ will skip all the services offered by " "the DIS_DNS server, while block=/SERVICE_LIST will skip all the " "SERVICE_LIST services offered by any server and DIS_DNS/SERVICE_LIST " "will skip DIS_DNS/SERVICE_LIST.\n" << endl; } // -------------------------------------------------------------------------- void SetupConfiguration(Configuration &conf) { const string n = conf.GetName()+".log"; po::options_description configp("Program options"); configp.add_options() ("dns", var("localhost"), "Dim nameserver host name (Overwites DIM_DNS_NODE environment variable)") ("log,l", var(n), "Write log-file") ("console,c", var(), "Use console (0=shell, 1=simple buffered, X=simple unbuffered)") ("cmd", vars(), "Execute one or more commands at startup") ("exec,e", vars(), "Execute one or more scrips at startup") ("quit", po_switch(), "Quit after startup"); ; po::options_description configs("DataLogger options"); configs.add_options() ("block,b", vars(), "Black-list to block services") ("allow,a", vars(), "White-list to only allowe certain services") ("debug,d", po_bool(), "Debug mode. Print clear text of received service reports.") ("group,g", vars(), "Grouping of services into a single run-Fits") ("run-timeout", var(), "Time out delay for old run numbers in milliseconds.") ("destination-folder", var(), "Base path for the nightly and run files") ("stats-interval", var(), "Interval in milliseconds for write statistics update") ("no-filename-service", po_bool(), "Disable update of filename service") ("no-numsubs-service", po_bool(), "Disable update of number-of-subscriptions service") ; conf.AddEnv("dns", "DIM_DNS_NODE"); conf.AddOptions(configp); conf.AddOptions(configs); } // -------------------------------------------------------------------------- int main(int argc, const char* argv[]) { Configuration conf(argv[0]); conf.SetPrintUsage(PrintUsage); SetupConfiguration(conf); po::variables_map vm; try { vm = conf.Parse(argc, argv); } #if BOOST_VERSION > 104000 catch (po::multiple_occurrences &e) { cerr << "Program options invalid due to: " << e.what() << " of '" << e.get_option_name() << "'." << endl; return -1; } #endif catch (exception& e) { cerr << "Program options invalid due to: " << e.what() << endl; return -1; } if (conf.HasVersion() || conf.HasPrint()) return -1; if (conf.HasHelp()) { PrintHelp(); return -1; } Dim::Setup(conf.Get("dns")); // try { // No console access at all if (!conf.Has("console")) return RunShell(conf); // Console access w/ and w/o Dim if (conf.Get("console")==0) return RunShell(conf); else return RunShell(conf); } /* catch (exception& e) { cerr << "Exception: " << e.what() << endl; return -1; }*/ return 0; }