File indexing completed on 2022-05-31 02:17:23
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
0042
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
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
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
0082
0083
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
0109 this->savePB(&*store, openHistoFilePathName, fp.run_, fp.lumi_);
0110
0111
0112 ::rename(openHistoFilePathName.c_str(), histoFilePathName.c_str());
0113 }
0114
0115 if (abortFlag)
0116 return;
0117
0118
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
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
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
0159 dataFileName = bfs::path(dataFilePathName).filename().string();
0160 }
0161
0162 bpt::ptree data;
0163 bpt::ptree processedEvents, acceptedEvents, errorEvents, bitmask, fileList, fileSize, inputFiles, fileAdler32,
0164 transferDestination, mergeType, hltErrorEvents;
0165
0166 processedEvents.put("", nProcessed);
0167 acceptedEvents.put("", nProcessed);
0168
0169 errorEvents.put("", 0);
0170 bitmask.put("", 0);
0171 fileList.put("", dataFileName);
0172 fileSize.put("", dataFileStat.st_size);
0173 inputFiles.put("", "");
0174 fileAdler32.put("", -1);
0175 transferDestination.put("", transferDestinationStr);
0176 mergeType.put("", mergeTypeStr);
0177 hltErrorEvents.put("", 0);
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
0197 bfs::path outJsonDefName{
0198 edm::Service<evf::EvFDaqDirector>()->baseRunDir()};
0199 outJsonDefName /= (std::string("output_") + oss_pid.str() + std::string(".jsd"));
0200 pt.put("definition", outJsonDefName.string());
0201 }
0202
0203 char sourceInfo[64];
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
0221
0222
0223
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
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
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
0268 for (const auto& qr : me->getQReports()) {
0269 std::string result;
0270
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
0286
0287 if (tag_ == "UNKNOWN") {
0288 qr_histo.set_streamed_histo((void const*)qr_buffer.Buffer(), qr_buffer.Length());
0289 } else {
0290
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
0299
0300
0301
0302
0303
0304
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
0319 gzip_stream.Close();
0320 file_stream.Close();
0321 } else {
0322
0323 dqmstore_message.SerializeToZeroCopyStream(&file_stream);
0324
0325
0326 file_stream.Close();
0327 }
0328
0329
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
0336
0337
0338 int n16kBlocks = (bufferSize + 16383) / 16384;
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;
0349 deflateStream.next_in = (Bytef*)buffer.Buffer();
0350 deflateStream.avail_out = (uInt)maxOutputSize;
0351 deflateStream.next_out = (Bytef*)compression_output;
0352
0353
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);