FileIO/plugins/DQMFileSaverPB.cc

0001 #include <filesystem>
0002 #include <fstream>
0003 #include <iostream>
0004 #include <string>
0005 #include <utility>
0006 #include <vector>
0007 #include <sys/stat.h>
0008 #include <sys/types.h>
0009 #include <unistd.h>
0010 #include <boost/property_tree/json_parser.hpp>
0011 #include <openssl/md5.h>
0012 #include <fmt/printf.h>
0013
0014 #include <google/protobuf/io/coded_stream.h>
0015 #include <google/protobuf/io/gzip_stream.h>
0016 #include <google/protobuf/io/zero_copy_stream_impl.h>
0017
0018 #include <TString.h>
0019 #include <TSystem.h>
0020 #include <TBufferFile.h>
0021
0022 #include "zlib.h"
0023 #include "DQMServices/Core/interface/DQMStore.h"
0024 #include "DQMServices/Core/interface/ROOTFilePB.pb.h"
0025 #include "FWCore/Framework/interface/LuminosityBlock.h"
0026 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0027 #include "FWCore/ServiceRegistry/interface/Service.h"
0028
0029 #include "DQMFileSaverPB.h"
0030
0031 using namespace dqm;
0032
0033 DQMFileSaverPB::DQMFileSaverPB(const edm::ParameterSet& ps) : DQMFileSaverBase(ps) {
0034   fakeFilterUnitMode_ = ps.getUntrackedParameter<bool>("fakeFilterUnitMode", false);
0035   streamLabel_ = ps.getUntrackedParameter<std::string>("streamLabel", "streamDQMHistograms");
0036   tag_ = ps.getUntrackedParameter<std::string>("tag", "UNKNOWN");
0037
0038   transferDestination_ = "";
0039   mergeType_ = "";
0040
0041   // If tag is set we're running in a DQM Live mode.
0042   // Snapshot files will be saved for every client, then they will be merged and uploaded to the new DQM GUI.
0043   if (tag_ != "UNKNOWN") {
0044     streamLabel_ = "DQMLive";
0045   }
0046 }
0047
0048 DQMFileSaverPB::~DQMFileSaverPB() = default;
0049
0050 void DQMFileSaverPB::initRun() const {
0051   if (!fakeFilterUnitMode_) {
0052     transferDestination_ = edm::Service<evf::EvFDaqDirector>()->getStreamDestinations(streamLabel_);
0053     mergeType_ = edm::Service<evf::EvFDaqDirector>()->getStreamMergeType(streamLabel_, evf::MergeTypePB);
0054   }
0055
0056   if (!fakeFilterUnitMode_) {
0057     evf::EvFDaqDirector* daqDirector = (evf::EvFDaqDirector*)(edm::Service<evf::EvFDaqDirector>().operator->());
0058     const std::string initFileName = daqDirector->getInitFilePath(streamLabel_);
0059     std::ofstream file(initFileName);
0060     file.close();
0061   }
0062 }
0063
0064 void DQMFileSaverPB::saveLumi(const FileParameters& fp) const {
0065   // get from DAQ2 services where to store the files according to their format
0066   namespace bpt = boost::property_tree;
0067
0068   std::string openJsonFilePathName;
0069   std::string jsonFilePathName;
0070   std::string openHistoFilePathName;
0071   std::string histoFilePathName;
0072
0073   evf::FastMonitoringService* fms = nullptr;
0074   edm::Service<DQMStore> store;
0075
0076   // create the files names
0077   if (fakeFilterUnitMode_) {
0078     std::string runDir = fmt::sprintf("%s/run%06d", fp.path_, fp.run_);
0079     std::string baseName = "";
0080     std::filesystem::create_directories(runDir);
0081     // If tag is configured, append it to the name of the resulting file.
0082     // This differentiates files saved by different clients.
0083     // If tag is not configured, we don't add it at all to keep the old behaviour unchanged.
0084     if (tag_ == "UNKNOWN") {
0085       baseName = fmt::sprintf("%s/run%06d_ls%04d_%s", runDir, fp.run_, fp.lumi_, streamLabel_);
0086     } else {
0087       baseName = fmt::sprintf("%s/run%06d_%s_%s", runDir, fp.run_, tag_, streamLabel_);
0088     }
0089
0090     jsonFilePathName = baseName + ".jsn";
0091     openJsonFilePathName = jsonFilePathName + ".open";
0092
0093     histoFilePathName = baseName + ".pb";
0094     openHistoFilePathName = histoFilePathName + ".open";
0095   } else {
0096     openJsonFilePathName = edm::Service<evf::EvFDaqDirector>()->getOpenOutputJsonFilePath(fp.lumi_, streamLabel_);
0097     jsonFilePathName = edm::Service<evf::EvFDaqDirector>()->getOutputJsonFilePath(fp.lumi_, streamLabel_);
0098
0099     openHistoFilePathName =
0100         edm::Service<evf::EvFDaqDirector>()->getOpenProtocolBufferHistogramFilePath(fp.lumi_, streamLabel_);
0101     histoFilePathName = edm::Service<evf::EvFDaqDirector>()->getProtocolBufferHistogramFilePath(fp.lumi_, streamLabel_);
0102
0103     fms = (evf::FastMonitoringService*)(edm::Service<evf::MicroStateService>().operator->());
0104   }
0105
0106   bool abortFlag = false;
0107   if (fms ? fms->getEventsProcessedForLumi(fp.lumi_, &abortFlag) : true) {
0108     // Save the file in the open directory.
0109     this->savePB(&*store, openHistoFilePathName, fp.run_, fp.lumi_);
0110
0111     // Now move the the data and json files into the output directory.
0112     ::rename(openHistoFilePathName.c_str(), histoFilePathName.c_str());
0113   }
0114
0115   if (abortFlag)
0116     return;
0117
0118   // Write the json file in the open directory.
0119   bpt::ptree pt = fillJson(fp.run_, fp.lumi_, histoFilePathName, transferDestination_, mergeType_, fms);
0120   write_json(openJsonFilePathName, pt);
0121   ::rename(openJsonFilePathName.c_str(), jsonFilePathName.c_str());
0122 }
0123
0124 void DQMFileSaverPB::saveRun(const FileParameters& fp) const {
0125   // no saving for the run
0126 }
0127
0128 boost::property_tree::ptree DQMFileSaverPB::fillJson(int run,
0129                                                      int lumi,
0130                                                      const std::string& dataFilePathName,
0131                                                      const std::string& transferDestinationStr,
0132                                                      const std::string& mergeTypeStr,
0133                                                      evf::FastMonitoringService* fms) {
0134   namespace bpt = boost::property_tree;
0135   namespace bfs = std::filesystem;
0136
0137   bpt::ptree pt;
0138
0139   int hostnameReturn;
0140   char host[32];
0141   hostnameReturn = gethostname(host, sizeof(host));
0142   if (hostnameReturn == -1)
0143     throw cms::Exception("fillJson") << "Internal error, cannot get host name";
0144
0145   int pid = getpid();
0146   std::ostringstream oss_pid;
0147   oss_pid << pid;
0148
0149   int nProcessed = fms ? (fms->getEventsProcessedForLumi(lumi)) : -1;
0150
0151   // Stat the data file: if not there, throw
0152   std::string dataFileName;
0153   struct stat dataFileStat;
0154   dataFileStat.st_size = 0;
0155   if (nProcessed) {
0156     if (stat(dataFilePathName.c_str(), &dataFileStat) != 0)
0157       throw cms::Exception("fillJson") << "Internal error, cannot get data file: " << dataFilePathName;
0158     // Extract only the data file name from the full path
0159     dataFileName = bfs::path(dataFilePathName).filename().string();
0160   }
0161   // The availability test of the FastMonitoringService was done in the ctor.
0162   bpt::ptree data;
0163   bpt::ptree processedEvents, acceptedEvents, errorEvents, bitmask, fileList, fileSize, inputFiles, fileAdler32,
0164       transferDestination, mergeType, hltErrorEvents;
0165
0166   processedEvents.put("", nProcessed);  // Processed events
0167   acceptedEvents.put("", nProcessed);   // Accepted events, same as processed for our purposes
0168
0169   errorEvents.put("", 0);                               // Error events
0170   bitmask.put("", 0);                                   // Bitmask of abs of CMSSW return code
0171   fileList.put("", dataFileName);                       // Data file the information refers to
0172   fileSize.put("", dataFileStat.st_size);               // Size in bytes of the data file
0173   inputFiles.put("", "");                               // We do not care about input files!
0174   fileAdler32.put("", -1);                              // placeholder to match output json definition
0175   transferDestination.put("", transferDestinationStr);  // SM Transfer destination field
0176   mergeType.put("", mergeTypeStr);                      // SM Transfer destination field
0177   hltErrorEvents.put("", 0);                            // Error events
0178
0179   data.push_back(std::make_pair("", processedEvents));
0180   data.push_back(std::make_pair("", acceptedEvents));
0181   data.push_back(std::make_pair("", errorEvents));
0182   data.push_back(std::make_pair("", bitmask));
0183   data.push_back(std::make_pair("", fileList));
0184   data.push_back(std::make_pair("", fileSize));
0185   data.push_back(std::make_pair("", inputFiles));
0186   data.push_back(std::make_pair("", fileAdler32));
0187   data.push_back(std::make_pair("", transferDestination));
0188   data.push_back(std::make_pair("", mergeType));
0189   data.push_back(std::make_pair("", hltErrorEvents));
0190
0191   pt.add_child("data", data);
0192
0193   if (fms == nullptr) {
0194     pt.put("definition", "/fakeDefinition.jsn");
0195   } else {
0196     // The availability test of the EvFDaqDirector Service was done in the ctor.
0197     bfs::path outJsonDefName{
0198         edm::Service<evf::EvFDaqDirector>()->baseRunDir()};  //we assume this file is written bu the EvF Output module
0199     outJsonDefName /= (std::string("output_") + oss_pid.str() + std::string(".jsd"));
0200     pt.put("definition", outJsonDefName.string());
0201   }
0202
0203   char sourceInfo[64];  //host and pid information
0204   sprintf(sourceInfo, "%s_%d", host, pid);
0205   pt.put("source", sourceInfo);
0206
0207   return pt;
0208 }
0209
0210 void DQMFileSaverPB::fillDescriptions(edm::ConfigurationDescriptions& descriptions) {
0211   edm::ParameterSetDescription desc;
0212   desc.setComment("Saves histograms from DQM store, HLT->pb workflow.");
0213
0214   desc.addUntracked<bool>("fakeFilterUnitMode", false)->setComment("If set, EvFDaqDirector is emulated and not used.");
0215
0216   desc.addUntracked<std::string>("streamLabel", "streamDQMHistograms")->setComment("Label of the stream.");
0217
0218   DQMFileSaverBase::fillDescription(desc);
0219
0220   // Changed to use addDefault instead of add here because previously
0221   // DQMFileSaverOnline and DQMFileSaverPB both used the module label
0222   // "saver" which caused conflicting cfi filenames to be generated.
0223   // add could be used if unique module labels were given.
0224   descriptions.addDefault(desc);
0225 }
0226
0227 void DQMFileSaverPB::savePB(DQMStore* store, std::string const& filename, int run, int lumi) const {
0228   using google::protobuf::io::FileOutputStream;
0229   using google::protobuf::io::GzipOutputStream;
0230   using google::protobuf::io::StringOutputStream;
0231
0232   unsigned int nme = 0;
0233
0234   dqmstorepb::ROOTFilePB dqmstore_message;
0235
0236   // We save all histograms, indifferent of the lumi flag: even tough we save per lumi, this is a *snapshot*.
0237   auto mes = store->getAllContents("");
0238   for (auto const me : mes) {
0239     TBufferFile buffer(TBufferFile::kWrite);
0240     if (me->kind() < MonitorElement::Kind::TH1F) {
0241       TObjString object(me->tagString().c_str());
0242       buffer.WriteObject(&object);
0243     } else {
0244       buffer.WriteObject(me->getRootObject());
0245     }
0246     dqmstorepb::ROOTFilePB::Histo& histo = *dqmstore_message.add_histo();
0247     histo.set_full_pathname(me->getFullname());
0248     uint32_t flags = 0;
0249     flags |= (uint32_t)me->kind();
0250     if (me->getLumiFlag())
0251       flags |= DQMNet::DQM_PROP_LUMI;
0252     if (me->getEfficiencyFlag())
0253       flags |= DQMNet::DQM_PROP_EFFICIENCY_PLOT;
0254     histo.set_flags(flags);
0255     histo.set_size(buffer.Length());
0256
0257     if (tag_ == "UNKNOWN") {
0258       histo.set_streamed_histo((void const*)buffer.Buffer(), buffer.Length());
0259     } else {
0260       // Compress ME blob with zlib
0261       int maxOutputSize = this->getMaxCompressedSize(buffer.Length());
0262       std::vector<char> compression_output(maxOutputSize);
0263       uLong total_out = this->compressME(buffer, maxOutputSize, compression_output.data());
0264       histo.set_streamed_histo(compression_output.data(), total_out);
0265     }
0266
0267     // Save quality reports
0268     for (const auto& qr : me->getQReports()) {
0269       std::string result;
0270       // TODO: 64 is likely too short; memory corruption in the old code?
0271       char buf[64];
0272       std::snprintf(buf, sizeof(buf), "qr=st:%d:%.*g:", qr->getStatus(), DBL_DIG + 2, qr->getQTresult());
0273       result = '<' + me->getName() + '.' + qr->getQRName() + '>';
0274       result += buf;
0275       result += qr->getAlgorithm() + ':' + qr->getMessage();
0276       result += "</" + me->getName() + '.' + qr->getQRName() + '>';
0277       TObjString str(result.c_str());
0278
0279       dqmstorepb::ROOTFilePB::Histo& qr_histo = *dqmstore_message.add_histo();
0280       TBufferFile qr_buffer(TBufferFile::kWrite);
0281       qr_buffer.WriteObject(&str);
0282       qr_histo.set_full_pathname(me->getFullname() + '.' + qr->getQRName());
0283       qr_histo.set_flags(static_cast<uint32_t>(MonitorElement::Kind::STRING));
0284       qr_histo.set_size(qr_buffer.Length());
0285       // qr_histo.set_streamed_histo((void const*)qr_buffer.Buffer(), qr_buffer.Length());
0286
0287       if (tag_ == "UNKNOWN") {
0288         qr_histo.set_streamed_histo((void const*)qr_buffer.Buffer(), qr_buffer.Length());
0289       } else {
0290         // Compress ME blob with zlib
0291         int maxOutputSize = this->getMaxCompressedSize(qr_buffer.Length());
0292         char compression_output[maxOutputSize];
0293         uLong total_out = this->compressME(qr_buffer, maxOutputSize, compression_output);
0294         qr_histo.set_streamed_histo(compression_output, total_out);
0295       }
0296     }
0297
0298     // Save efficiency tag, if any.
0299     // XXX not supported by protobuf files.
0300
0301     // Save tag if any.
0302     // XXX not supported by protobuf files.
0303
0304     // Count saved histograms
0305     ++nme;
0306   }
0307
0308   int filedescriptor =
0309       ::open(filename.c_str(), O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH);
0310   FileOutputStream file_stream(filedescriptor);
0311   if (tag_ == "UNKNOWN") {
0312     GzipOutputStream::Options options;
0313     options.format = GzipOutputStream::GZIP;
0314     options.compression_level = 1;
0315     GzipOutputStream gzip_stream(&file_stream, options);
0316     dqmstore_message.SerializeToZeroCopyStream(&gzip_stream);
0317
0318     // Flush the internal streams & Close the file descriptor
0319     gzip_stream.Close();
0320     file_stream.Close();
0321   } else {
0322     // We zlib compressed individual MEs so no need to compress the entire file again.
0323     dqmstore_message.SerializeToZeroCopyStream(&file_stream);
0324
0325     // Flush the internal stream & Close the file descriptor
0326     file_stream.Close();
0327   }
0328
0329   // Maybe make some noise.
0330   edm::LogInfo("DQMFileSaverPB") << "savePB: successfully wrote " << nme << " objects  "
0331                                  << "into DQM file '" << filename << "'\n";
0332 }
0333
0334 int DQMFileSaverPB::getMaxCompressedSize(int bufferSize) const {
0335   // When input data is very badly compressable, zlib will add overhead instead of reducing the size.
0336   // There is a minor amount of overhead (6 bytes overall and 5 bytes per 16K block) that is taken
0337   // into consideration here to find out potential absolute maximum size of the output.
0338   int n16kBlocks = (bufferSize + 16383) / 16384;  // round up any fraction of a block
0339   int maxOutputSize = bufferSize + 6 + (n16kBlocks * 5);
0340   return maxOutputSize;
0341 }
0342
0343 ulong DQMFileSaverPB::compressME(const TBufferFile& buffer, int maxOutputSize, char* compression_output) const {
0344   z_stream deflateStream;
0345   deflateStream.zalloc = Z_NULL;
0346   deflateStream.zfree = Z_NULL;
0347   deflateStream.opaque = Z_NULL;
0348   deflateStream.avail_in = (uInt)buffer.Length() + 1;   // size of input, string + terminator
0349   deflateStream.next_in = (Bytef*)buffer.Buffer();      // input array
0350   deflateStream.avail_out = (uInt)maxOutputSize;        // size of output
0351   deflateStream.next_out = (Bytef*)compression_output;  // output array, result will be placed here
0352
0353   // The actual compression
0354   deflateInit(&deflateStream, Z_BEST_COMPRESSION);
0355   deflate(&deflateStream, Z_FINISH);
0356   deflateEnd(&deflateStream);
0357
0358   return deflateStream.total_out;
0359 }
0360
0361 #include "FWCore/Framework/interface/MakerMacros.h"
0362 DEFINE_FWK_MODULE(DQMFileSaverPB);