Commit 157011e

if (NOT CMAKE_SYSTEM_NAME STREQUAL "Darwin")

o2_add_executable(epn-stderr-monitor

COMPONENT_NAME epn

SOURCES src/EPNstderrMonitor.cxx

PUBLIC_LINK_LIBRARIES FairMQ::FairMQ AliceO2::InfoLogger)

endif()

#include <fairmq/Device.h>

#include <fairmq/runDevice.h>

#include "InfoLogger/InfoLogger.hxx"

#include <string>

#include <thread>

#include <vector>

#include <unordered_map>

#include <regex>

#include <filesystem>

#include <chrono>

#include <fstream>

#include <unistd.h>

#include <sys/inotify.h>

#include <poll.h>

using namespace AliceO2;

static constexpr size_t MAX_LINES_FILE = 30;

static constexpr size_t MAX_BYTES_FILE = MAX_LINES_FILE * 512;

static constexpr size_t MAX_LINES_TOTAL = 1000;

static constexpr size_t MAX_BYTES_TOTAL = MAX_LINES_TOTAL * 256;

struct fileMon {

std::ifstream file;

std::string name;

unsigned int nLines = 0;

unsigned int nBytes = 0;

fileMon(const std::string& path, const std::string& filename);

};

fileMon::fileMon(const std::string& path, const std::string& filename)

{

printf("Monitoring file %s\n", filename.c_str());

name = filename;

file.open(path + "/" + filename, std::ifstream::in);

}

class EPNMonitor

{

public:

EPNMonitor(std::string path, bool infoLogger, int runNumber, std::string partition);

~EPNMonitor();

private:

void thread();

void check_add_file(const std::string& filename);

void sendLog(const std::string& file, const std::string& message);

bool mInfoLoggerActive;

volatile bool mTerminate = false;

std::thread mThread;

std::unordered_map<std::string, fileMon> mFiles;

std::string mPath;

std::vector<std::regex> mFilters;

unsigned int mRunNUmber;

std::string mPartition;

unsigned int nLines = 0;

unsigned int nBytes = 0;

std::unique_ptr<InfoLogger::InfoLogger> mLogger;

std::unique_ptr<InfoLogger::InfoLoggerContext> mLoggerContext;

};

EPNMonitor::EPNMonitor(std::string path, bool infoLogger, int runNumber, std::string partition)

{

mFilters.emplace_back("^Info in <");

mFilters.emplace_back("^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{6}");

mInfoLoggerActive = infoLogger;

mPath = path;

mRunNUmber = runNumber;

mPartition = partition;

if (infoLogger) {

mLogger = std::make_unique<InfoLogger::InfoLogger>();

mLoggerContext = std::make_unique<InfoLogger::InfoLoggerContext>();

mLoggerContext->setField(InfoLogger::InfoLoggerContext::FieldName::Partition, partition != "" ? partition : "unspecified");

mLoggerContext->setField(InfoLogger::InfoLoggerContext::FieldName::Run, runNumber == 0 ? std::to_string(runNumber) : "unspecified");

}

mThread = std::thread(&EPNMonitor::thread, this);

}

EPNMonitor::~EPNMonitor()

{

mTerminate = true;

mThread.join();

}

void EPNMonitor::check_add_file(const std::string& filename)

{

//printf("Checking '%s'\n", filename.c_str());

static const std::regex match_stderr("_err\\.log$");

if (std::regex_search(filename, match_stderr)) {

mFiles.try_emplace(filename, mPath, filename);

}

}

void EPNMonitor::sendLog(const std::string& file, const std::string& message)

{

if (mInfoLoggerActive) {

mLoggerContext->setField(InfoLogger::InfoLoggerContext::FieldName::Facility, "stderr/" + file);

static const InfoLogger::InfoLogger::InfoLoggerMessageOption opt = {InfoLogger::InfoLogger::Severity::Error, 3, InfoLogger::InfoLogger::undefinedMessageOption.errorCode, InfoLogger::InfoLogger::undefinedMessageOption.sourceFile, InfoLogger::InfoLogger::undefinedMessageOption.sourceLine};

mLogger->log(opt, *mLoggerContext, "stderr: %s", message.c_str());

} else {

printf("stderr: %s: %s\n", file.c_str(), message.c_str());

}

}

void EPNMonitor::thread()

{

printf("EPN stderr Monitor active\n");

int fd;

int wd;

static constexpr size_t BUFFER_SIZE = 64 * 1024;

std::vector<char> evt_buffer(BUFFER_SIZE);

std::vector<char> text_buffer(8192);

fd = inotify_init();

wd = inotify_add_watch(fd, mPath.c_str(), IN_CREATE);

if (fd < 0) {

throw std::runtime_error(std::string("Error initializing inotify ") + std::to_string(fd) + " " + std::to_string(wd));

}

pollfd pfd = {fd, POLLIN, 0};

for (const auto& entry : std::filesystem::directory_iterator(mPath)) {

if (entry.is_regular_file()) {

check_add_file(entry.path().filename());

}

}

auto lastTime = std::chrono::system_clock::now();

while (!mTerminate) {

if (poll(&pfd, 1, 50) > 0) {

int l = read(fd, evt_buffer.data(), BUFFER_SIZE);

if (l < 0) {

throw std::runtime_error(std::string("Error waiting for inotify event ") + std::to_string(l));

}

for (int i = 0; i < l; i += sizeof(inotify_event)) {

inotify_event* event = (inotify_event*)&evt_buffer[i];

if (event->len && (event->mask & IN_CREATE) && !(event->mask & IN_ISDIR)) {

check_add_file(event->name);

}

i += event->len;

}

}

auto curTime = std::chrono::system_clock::now();

if (std::chrono::duration_cast<std::chrono::milliseconds>(curTime - lastTime).count() >= 1000) {

char* ptr = text_buffer.data();

std::string line;

for (auto fit = mFiles.begin(); fit != mFiles.end(); fit++) {

auto& f = fit->second;

auto& file = f.file;

file.clear();

do {

std::getline(file, line);

if (line.size()) {

bool filterLine = false;

for (const auto& filter : mFilters) {

if (std::regex_search(line, filter)) {

filterLine = true;

break;

}

}

if (filterLine) {

continue;

}

f.nLines++;

f.nBytes += line.size();

nLines++;

nBytes += line.size();

if (f.nLines >= MAX_LINES_FILE || f.nBytes >= MAX_BYTES_FILE) {

sendLog(f.name, "Exceeded log size for process " + f.name + " (" + std::to_string(f.nLines) + " lines, " + std::to_string(f.nBytes) + " bytes), not reporting any more errors from this file...");

fit = mFiles.erase(fit);

break;

}

if (nLines >= MAX_LINES_TOTAL || nBytes >= MAX_BYTES_TOTAL) {

break;

}

sendLog(f.name, line);

}

} while (!file.eof());

}

lastTime = curTime;

}

if (nLines >= MAX_LINES_TOTAL || nBytes >= MAX_BYTES_TOTAL) {

sendLog("", "Max total stderr log size exceeded (" + std::to_string(nLines) + " lines, " + std::to_string(nBytes) + "), not sending any more stderr logs from this node...");

break;

}

usleep(50000);

}

inotify_rm_watch(fd, wd);

close(fd);

printf("EPN stderr Monitor terminating\n");

}

static std::unique_ptr<EPNMonitor> gEPNMonitor;

namespace bpo = boost::program_options;

struct EPNstderrMonitor : fair::mq::Device {

void InitTask() override

{

std::string path = ".";

bool infoLogger = fConfig->GetProperty<int>("infologger");

bool dds = false;

if (fConfig->Count("plugin")) {

const auto& plugins = fConfig->GetProperty<std::vector<std::string>>("plugin");

bool dds = std::find(plugins.begin(), plugins.end(), "ODC") != plugins.end();

}

bool runNumber = dds ? fConfig->GetProperty<int>("runNumber") : 0;

std::string partition = "";

gEPNMonitor = std::make_unique<EPNMonitor>(path, infoLogger, runNumber, partition);

}

bool ConditionalRun() override

{

return true;

}

};

void addCustomOptions(bpo::options_description& options)

{

options.add_options()("infologger", bpo::value<int>()->default_value(0), "Send via infologger");

}

std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& config)

{

return std::make_unique<EPNstderrMonitor>();

}