StreamerIO/plugins/RamdiskMonitor.cc

0001 #include <exception>
0002 #include <filesystem>
0003 #include <map>
0004 #include <memory>
0005 #include <string>
0006 #include <sys/stat.h>
0007 #include <vector>
0008
0009 #include <boost/algorithm/string/predicate.hpp>
0010 #include <boost/range.hpp>
0011 #include <boost/regex.hpp>
0012
0013 #include <fmt/printf.h>
0014
0015 #include "DQMServices/Core/interface/DQMOneEDAnalyzer.h"
0016 #include "DQMServices/Core/interface/DQMStore.h"
0017 #include "FWCore/Framework/interface/Event.h"
0018 #include "FWCore/Framework/interface/EventSetup.h"
0019 #include "FWCore/MessageLogger/interface/MessageLogger.h"
0020 #include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
0021 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0022 #include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
0023 #include "DQMFileIterator.h"
0024
0025 namespace dqm {
0026
0027   namespace rdm {
0028     struct Empty {};
0029   }  // namespace rdm
0030
0031   class RamdiskMonitor : public DQMOneEDAnalyzer<edm::LuminosityBlockCache<rdm::Empty>> {
0032   public:
0033     RamdiskMonitor(const edm::ParameterSet &ps);
0034     ~RamdiskMonitor() override = default;
0035     static void fillDescriptions(edm::ConfigurationDescriptions &descriptions);
0036
0037   protected:
0038     void bookHistograms(DQMStore::IBooker &, edm::Run const &, edm::EventSetup const &) override;
0039     std::shared_ptr<rdm::Empty> globalBeginLuminosityBlock(edm::LuminosityBlock const &lumi,
0040                                                            edm::EventSetup const &eSetup) const override;
0041     void globalEndLuminosityBlock(edm::LuminosityBlock const &lumi, edm::EventSetup const &eSetup) final {}
0042     void analyze(edm::Event const &e, edm::EventSetup const &eSetup) override {}
0043
0044     void analyzeFile(std::string fn, unsigned int run, unsigned int lumi, std::string label) const;
0045     double getRunTimestamp() const;
0046
0047     const unsigned int runNumber_;
0048     const std::string runInputDir_;
0049     const std::vector<std::string> streamLabels_;
0050     const std::string runPath_;
0051
0052     struct StreamME {
0053       MonitorElement *eventsAccepted;
0054       MonitorElement *eventsProcessed;
0055
0056       MonitorElement *deliveryDelayMTime;
0057       MonitorElement *deliveryDelayCTime;
0058     };
0059
0060     std::map<std::string, StreamME> streams_;
0061     mutable std::set<std::string> filesSeen_;
0062     mutable double global_start_ = 0.;
0063
0064     static constexpr double LUMI = 23.310893056;
0065   };
0066
0067   RamdiskMonitor::RamdiskMonitor(const edm::ParameterSet &ps)
0068       : runNumber_{ps.getUntrackedParameter<unsigned int>("runNumber")},
0069         runInputDir_{ps.getUntrackedParameter<std::string>("runInputDir")},
0070         streamLabels_{ps.getUntrackedParameter<std::vector<std::string>>("streamLabels")},
0071         runPath_{fmt::sprintf("%s/run%06d", runInputDir_, runNumber_)} {}
0072
0073   void RamdiskMonitor::bookHistograms(DQMStore::IBooker &ib, edm::Run const &, edm::EventSetup const &) {
0074     for (const auto &stream : streamLabels_) {
0075       edm::LogInfo("RamdiskMonitor") << "Booking: " << stream;
0076
0077       ib.cd();
0078       ib.setCurrentFolder(std::string("Info/RamdiskMonitor/") + stream + "/");
0079
0080       StreamME m;
0081
0082       m.eventsAccepted = ib.book1D("EventAccepted", "# of accepted events per lumi", 4, 0., 4.);
0083       m.eventsProcessed = ib.book1D("EventProcessed", "# of processed events per lumi", 4, 0., 4.);
0084       m.deliveryDelayMTime =
0085           ib.book1D("DeliveryDelayMTime", "Observed delivery delay for the data file (mtime).", 4, 0., 4.);
0086       m.deliveryDelayCTime =
0087           ib.book1D("DeliveryDelayCTime", "Observed delivery delay for the data file (ctime).", 4, 0., 4.);
0088
0089       m.eventsAccepted->getTH1F()->SetCanExtend(TH1::kXaxis);
0090       m.eventsProcessed->getTH1F()->SetCanExtend(TH1::kXaxis);
0091       m.deliveryDelayMTime->getTH1F()->SetCanExtend(TH1::kXaxis);
0092       m.deliveryDelayCTime->getTH1F()->SetCanExtend(TH1::kXaxis);
0093
0094       m.eventsAccepted->setAxisTitle("Luminosity Section", 1);
0095       m.eventsProcessed->setAxisTitle("Luminosity Section", 1);
0096       m.deliveryDelayMTime->setAxisTitle("Luminosity Section", 1);
0097       m.deliveryDelayCTime->setAxisTitle("Luminosity Section", 1);
0098
0099       m.eventsAccepted->setAxisTitle("Number of events", 2);
0100       m.eventsProcessed->setAxisTitle("Number of events", 2);
0101       m.deliveryDelayMTime->setAxisTitle("Delay (s.)", 2);
0102       m.deliveryDelayCTime->setAxisTitle("Delay (s.)", 2);
0103
0104       streams_[stream] = m;
0105     }
0106   };
0107
0108   double RamdiskMonitor::getRunTimestamp() const {
0109     if (global_start_ != 0)
0110       return global_start_;
0111
0112     std::string run_global = fmt::sprintf("%s/.run%06d.global", runInputDir_, runNumber_);
0113     struct stat st;
0114     if (::stat(run_global.c_str(), &st) != 0) {
0115       edm::LogWarning("RamdiskMonitor") << "Stat failed: " << run_global;
0116       return 0.;
0117     }
0118
0119     global_start_ = st.st_mtime;
0120     edm::LogPrint("RamdiskMonitor") << "Run start timestamp: " << global_start_;
0121     return global_start_;
0122   };
0123
0124   void RamdiskMonitor::analyzeFile(std::string fn, unsigned int run, unsigned int lumi, std::string label) const {
0125     using LumiEntry = dqmservices::DQMFileIterator::LumiEntry;
0126
0127     // we are disabled, at least for this stream
0128     if (streams_.empty())
0129       return;
0130
0131     auto itStream = streams_.find(label);
0132     if (itStream == streams_.end()) {
0133       edm::LogPrint("RamdiskMonitor") << "Stream not monitored [" << label << "]: " << fn;
0134       return;
0135     }
0136
0137     StreamME m = itStream->second;
0138
0139     // decode json and fill in some histograms
0140     LumiEntry lumi_jsn = LumiEntry::load_json(runPath_, fn, lumi, -1);
0141     m.eventsAccepted->setBinContent(lumi, lumi_jsn.n_events_accepted);
0142     m.eventsProcessed->setBinContent(lumi, lumi_jsn.n_events_processed);
0143
0144     // collect stat struct and calculate mtimes
0145     struct stat st;
0146     if (::stat(fn.c_str(), &st) != 0) {
0147       edm::LogWarning("RamdiskMonitor") << "Stat failed: " << fn;
0148       return;
0149     }
0150
0151     // get start offset (from .global)
0152     // abort the calculation if it does not exist
0153     double start_offset = getRunTimestamp();
0154     if (start_offset <= 0)
0155       return;
0156
0157     // check fff_dqmtools (separate repository)
0158     // for calculation details
0159     double mtime = st.st_mtime;
0160     double ctime = st.st_ctime;
0161
0162     // timeout from the begging of the run
0163     double start_offset_mtime = mtime - start_offset - LUMI;
0164     double start_offset_ctime = ctime - start_offset - LUMI;
0165     double lumi_offset = (lumi - 1) * LUMI;
0166
0167     // timeout from the time we think this lumi happenned
0168     double delay_mtime = start_offset_mtime - lumi_offset;
0169     double delay_ctime = start_offset_ctime - lumi_offset;
0170
0171     m.deliveryDelayMTime->setBinContent(lumi, delay_mtime);
0172     m.deliveryDelayCTime->setBinContent(lumi, delay_ctime);
0173   };
0174
0175   std::shared_ptr<dqm::rdm::Empty> RamdiskMonitor::globalBeginLuminosityBlock(edm::LuminosityBlock const &,
0176                                                                               edm::EventSetup const &eSetup) const {
0177     // search filesystem to find available lumi section files
0178     using std::filesystem::directory_entry;
0179     using std::filesystem::directory_iterator;
0180
0181     directory_iterator dend;
0182     for (directory_iterator di(runPath_); di != dend; ++di) {
0183       const boost::regex fn_re("run(\\d+)_ls(\\d+)_([a-zA-Z0-9]+)(_.*)?\\.jsn");
0184
0185       const std::string filename = di->path().filename().string();
0186       const std::string fn = di->path().string();
0187
0188       if (filesSeen_.find(filename) != filesSeen_.end()) {
0189         continue;
0190       }
0191
0192       boost::smatch result;
0193       if (boost::regex_match(filename, result, fn_re)) {
0194         unsigned int run = std::stoi(result[1]);
0195         unsigned int lumi = std::stoi(result[2]);
0196         std::string label = result[3];
0197
0198         filesSeen_.insert(filename);
0199
0200         if (run != runNumber_)
0201           continue;
0202
0203         // check if this is EoR
0204         if ((lumi == 0) && (label == "EoR")) {
0205           // do not handle
0206           continue;
0207         }
0208
0209         try {
0210           this->analyzeFile(fn, run, lumi, label);
0211         } catch (const std::exception &e) {
0212           // it's likely we have read it too soon
0213           filesSeen_.erase(filename);
0214
0215           std::string msg("Found, tried to load the json, but failed (");
0216           msg += e.what();
0217           msg += "): ";
0218           edm::LogWarning("RamdiskMonitor") << msg;
0219         }
0220       }
0221     }
0222
0223     // @TODO lookup info for the current lumi
0224     return std::shared_ptr<dqm::rdm::Empty>();
0225   }
0226
0227   void RamdiskMonitor::fillDescriptions(edm::ConfigurationDescriptions &d) {
0228     edm::ParameterSetDescription desc;
0229
0230     desc.setComment(
0231         "Analyses file timestams in the /fff/ramdisk and creates monitor "
0232         "elements.");
0233
0234     desc.addUntracked<std::vector<std::string>>("streamLabels")->setComment("List of streams to monitor.");
0235
0236     desc.addUntracked<unsigned int>("runNumber")->setComment("Run number passed via configuration file.");
0237
0238     desc.addUntracked<std::string>("runInputDir")->setComment("Directory where the DQM files will appear.");
0239
0240     d.add("RamdiskMonitor", desc);
0241   }
0242
0243 }  // namespace dqm
0244
0245 #include "FWCore/Framework/interface/MakerMacros.h"
0246 typedef dqm::RamdiskMonitor RamdiskMonitor;
0247 DEFINE_FWK_MODULE(RamdiskMonitor);