QPR-10445 -- Add EventMessage and FileIO wrapper class that adds a retry logic to some commen file IO operations

Author: NathanielVolfango

OREAnalytics/orea/app/inputparameters.hpp

@@ -267,6 +267,8 @@ class InputParameters {
     const boost::shared_ptr<ore::data::EngineData>& pricingEngine() { return pricingEngine_; }
     const boost::shared_ptr<ore::data::TodaysMarketParameters>& todaysMarketParams() { return todaysMarketParams_; }
     const boost::shared_ptr<ore::data::Portfolio>& portfolio() { return portfolio_; }
+
+    QuantLib::Size maxRetries() const { return maxRetries_; }
     QuantLib::Size nThreads() const { return nThreads_; }
     bool entireMarket() { return entireMarket_; }
     bool allFixings() { return allFixings_; }
@@ -471,7 +473,9 @@ class InputParameters {
     boost::shared_ptr<ore::data::EngineData> pricingEngine_;
     boost::shared_ptr<ore::data::TodaysMarketParameters> todaysMarketParams_;
     boost::shared_ptr<ore::data::Portfolio> portfolio_;
-    QuantLib::Size nThreads_ = 1;   
+    QuantLib::Size maxRetries_ = 7;
+    QuantLib::Size nThreads_ = 1;
+   
     bool entireMarket_ = false; 
     bool allFixings_ = false; 
     bool eomInflationFixings_ = true;

OREData/ored/CMakeLists.txt

@@ -259,6 +259,7 @@ utilities/currencyconfig.cpp
 utilities/currencyhedgedequityindexdecomposition.cpp
 utilities/currencyparser.cpp
 utilities/dategrid.cpp
+utilities/fileio.cpp
 utilities/filteredbufferedlogger.cpp
 utilities/flowanalysis.cpp
 utilities/indexnametranslator.cpp
@@ -562,6 +563,7 @@ utilities/currencyconfig.hpp
 utilities/currencyhedgedequityindexdecomposition.hpp
 utilities/currencyparser.hpp
 utilities/dategrid.hpp
+utilities/fileio.hpp
 utilities/filteredbufferedlogger.hpp
 utilities/flowanalysis.hpp
 utilities/indexnametranslator.hpp

OREData/ored/ored.hpp

@@ -289,6 +289,7 @@
 #include <ored/utilities/currencyhedgedequityindexdecomposition.hpp>
 #include <ored/utilities/currencyparser.hpp>
 #include <ored/utilities/dategrid.hpp>
+#include <ored/utilities/fileio.hpp>
 #include <ored/utilities/filteredbufferedlogger.hpp>
 #include <ored/utilities/flowanalysis.hpp>
 #include <ored/utilities/indexnametranslator.hpp>

OREData/ored/report/csvreport.cpp

@@ -20,6 +20,7 @@
 #include <boost/filesystem/path.hpp>
 #include <boost/filesystem/operations.hpp>
 #include <ored/report/csvreport.hpp>
+#include <ored/utilities/fileio.hpp>
 #include <ored/utilities/to_string.hpp>
 #include <ql/errors.hpp>
 #include <ql/math/comparison.hpp>
@@ -85,7 +86,8 @@ class ReportTypePrinter : public boost::static_visitor<> {
 };
 
 CSVFileReport::CSVFileReport(const string& filename, const char sep, const bool commentCharacter, char quoteChar,
-                             const string& nullString, bool lowerHeader, QuantLib::Size rolloverSize)
+                             const string& nullString, bool lowerHeader,
+                             QuantLib::Size rolloverSize)
     : filename_(filename), sep_(sep), commentCharacter_(commentCharacter), quoteChar_(quoteChar),
       nullString_(nullString), lowerHeader_(lowerHeader), rolloverSize_(rolloverSize), i_(0), fp_(NULL) {    
     baseFilename_ = filename_;
@@ -101,7 +103,7 @@ CSVFileReport::~CSVFileReport() {
 
 void CSVFileReport::open() {
     LOG("Opening CSV file report '" << filename_ << "'");
-    fp_ = fopen(filename_.c_str(), "w");
+    fp_ = FileIO::fopen(filename_.c_str(), "w");
     QL_REQUIRE(fp_, "Error opening file '" << filename_ << "'");
     finalized_ = false;
 }

OREData/ored/utilities/fileio.cpp

@@ -0,0 +1,99 @@
+/*
+ Copyright (C) 2023 Quaternion Risk Management Ltd
+ All rights reserved.
+
+ This file is part of ORE, a free-software/open-source library
+ for transparent pricing and risk analysis - http://opensourcerisk.org
+
+ ORE is free software: you can redistribute it and/or modify it
+ under the terms of the Modified BSD License.  You should have received a
+ copy of the license along with this program.
+ The license is also available online at <http://opensourcerisk.org>
+
+ This program is distributed on the basis that it will form a useful
+ contribution to risk analytics and model standardisation, but WITHOUT
+ ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ FITNESS FOR A PARTICULAR PURPOSE. See the license for more details.
+*/
+
+/*! \file ored/utilties/fileio.cpp
+    \brief Wrapper class for retrying file IO operations
+    \ingroup
+*/
+
+#include <boost/filesystem/operations.hpp>
+#include <chrono>
+#include <ored/utilities/fileio.hpp>
+#include <ored/utilities/log.hpp>
+#include <ored/utilities/to_string.hpp>
+#include <thread>
+#include <vector>
+
+namespace ore {
+namespace data {
+
+using QuantLib::Real;
+using QuantLib::Size;
+
+// Defaults
+static Size _s_maxRetries = 7;
+static std::vector<Real> _s_backoff = {0.5, 1, 2, 4, 8, 16, 30};
+
+Size FileIO::maxRetries() { return _s_maxRetries; }
+
+void FileIO::setMaxRetries(Size numOfRetries) {
+    LOG("Setting FileOpen max retries to " << numOfRetries);
+    _s_maxRetries = numOfRetries;
+}
+
+FILE* FileIO::fopen(const char* filename, const char* mode) {
+    FILE* fp;
+    int i = 0;
+
+    do {
+        if (i > 0) {
+            Real backoff = (i >= _s_backoff.size()) ? _s_backoff.back() : _s_backoff[i - 1];
+            int backoffMillis = backoff * 1000;
+            auto em = EventMessage("Error opening file '" + std::string(filename) + "'.");
+            em.set("retry_count", i);
+            em.set("retry_interval", backoffMillis);
+            WLOG(em);
+            std::this_thread::sleep_for(std::chrono::milliseconds(backoffMillis));
+        }
+
+        fp = std::fopen(filename, mode);
+
+        i++;
+    } while (fp == nullptr && i <= maxRetries());
+
+    return fp;
+}
+
+bool FileIO::create_directories(const path& p) {
+    bool res = false;
+    int i = 0;
+
+    do {
+        if (i > 0) {
+            Real backoff = (i >= _s_backoff.size()) ? _s_backoff.back() : _s_backoff[i - 1];
+            int backoffMillis = backoff * 1000;
+            auto em = EventMessage("Error creating directory '" + p.string() + "'.");
+            em.set("retry_count", i);
+            em.set("retry_interval", backoffMillis);
+            WLOG(em);
+            std::this_thread::sleep_for(std::chrono::milliseconds(backoffMillis));
+        }
+
+        try {
+            res = boost::filesystem::create_directories(p);
+        } catch (...) {
+        }
+
+        i++;
+    } while (res == false && i <= maxRetries());
+
+    return res;
+}
+
+} // namespace data
+} // namespace ore

OREData/ored/utilities/fileio.hpp

@@ -0,0 +1,52 @@
+/*
+ Copyright (C) 2023 Quaternion Risk Management Ltd
+ All rights reserved.
+
+ This file is part of ORE, a free-software/open-source library
+ for transparent pricing and risk analysis - http://opensourcerisk.org
+
+ ORE is free software: you can redistribute it and/or modify it
+ under the terms of the Modified BSD License.  You should have received a
+ copy of the license along with this program.
+ The license is also available online at <http://opensourcerisk.org>
+
+ This program is distributed on the basis that it will form a useful
+ contribution to risk analytics and model standardisation, but WITHOUT
+ ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ FITNESS FOR A PARTICULAR PURPOSE. See the license for more details.
+*/
+
+/*! \file ored/utilities/fileio.hpp
+    \brief Wrapper class for retrying file IO operations
+    \ingroup utilities
+*/
+
+#pragma once
+
+#include <boost/filesystem/path.hpp>
+#include <ql/types.hpp>
+#include <stdio.h>
+
+namespace ore {
+namespace data {
+
+using boost::filesystem::path;
+
+class FileIO {
+
+public:
+    FileIO() = delete;
+
+    //! The maximum number of retries, defaults to 7
+    static QuantLib::Size maxRetries();
+    static void setMaxRetries(QuantLib::Size);
+
+    //! Retry wrapper for std::fopen
+    static FILE* fopen(const char*, const char*);
+
+    //! Retry wrapper for boost::filesystem::create_directories
+    static bool create_directories(const path&);
+};
+
+} // namespace data
+} // namespace ore

OREData/ored/utilities/filteredbufferedlogger.cpp

@@ -41,8 +41,8 @@ void FilteredBufferedLogger::log(unsigned lvl, const std::string& log) {
     if (n != string::npos) {
         // Strip it out
         string log_message = log.substr(n + strlen(ore::data::StructuredMessage::name));
-	// check if we have already logged the mssage
-	std::size_t message_hash = std::hash<std::string>{}(log_message);
+        // check if we have already logged the mssage
+        std::size_t message_hash = std::hash<std::string>{}(log_message);
         if (message_hash_history_.insert(message_hash).second) {
             // log it
             BufferLogger::log(lvl, log_message);

OREData/ored/utilities/log.cpp

@@ -24,6 +24,7 @@
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include <iomanip>
 #include <ored/utilities/log.hpp>
+#include <ored/utilities/to_string.hpp>
 #include <ql/errors.hpp>
 
 using namespace boost::posix_time;
@@ -219,11 +220,11 @@ LoggerStream::~LoggerStream() {
 }
 
 string StructuredMessage::json() const {
-    string msg =
-        "{ \"category\":\"" + category_ + "\", \"group\":\"" + group_ + "\"," + " \"message\":\"" + jsonify(message_);
+    string msg = "{ \"category\":\"" + category_ + "\", \"group\":\"" + group_ + "\"," + " \"message\":\"" +
+                 jsonify(message_) + "\"";
 
     if (!subFields_.empty()) {
-        msg += "\", \"sub_fields\": [ ";
+        msg += ", \"sub_fields\": [ ";
         QuantLib::Size i = 0;
         for (const auto& p : subFields_) {
             // Only include subFields that are non-empty.
@@ -249,5 +250,50 @@ string StructuredMessage::jsonify(const string& s) const {
     boost::replace_all(str, "\n", "\\n");
     return str;
 }
+
+string EventMessage::json() const {
+    string msg = "{ \"exception_message\":\"" + jsonify(message_) + "\"";
+
+    if (!data_.empty()) {
+        msg += ", ";
+        QuantLib::Size i = 0;
+        for (const auto& p : data_) {
+            string value;
+            if (p.second.type() == typeid(string)) {
+                value = "\"" + boost::any_cast<string>(p.second) + "\"";
+            } else if (p.second.type() == typeid(boost::posix_time::ptime)) {
+                auto time = boost::any_cast<boost::posix_time::ptime>(p.second);
+                value = boost::posix_time::to_iso_extended_string(time);
+            } else if (p.second.type() == typeid(int)) {
+                value = to_string(boost::any_cast<int>(p.second));
+            } else if (p.second.type() == typeid(QuantLib::Real)) {
+                value = to_string(boost::any_cast<QuantLib::Real>(p.second));
+            } else if (p.second.type() == typeid(bool)) {
+                value = to_string(boost::any_cast<bool>(p.second));
+            } else {
+                WLOG(StructuredMessage("Error", "Event Logging", "Unrecognised value type for key '" + p.first + "'",
+                                       std::pair<string, string>()));
+            }
+
+            if (i > 0)
+                msg += ", ";
+            msg += "\"" + p.first + "\": " + value;
+            i++;
+        }
+    }
+    msg += " }";
+
+    return msg;
+}
+
+string EventMessage::jsonify(const string& s) const {
+    string str = s;
+    boost::replace_all(str, "\\", "\\\\"); // do this before the below otherwise we get \\"
+    boost::replace_all(str, "\"", "\\\"");
+    boost::replace_all(str, "\r", "\\r");
+    boost::replace_all(str, "\n", "\\n");
+    return str;
+}
+
 } // namespace data
 } // namespace ore

OREData/ored/utilities/log.hpp

@@ -53,6 +53,7 @@
 #include <ql/patterns/singleton.hpp>
 #include <sstream>
 
+#include <boost/any.hpp>
 #include <boost/thread/shared_mutex.hpp>
 #include <boost/thread/lock_types.hpp>
 
@@ -478,6 +479,30 @@ class StructuredMessage {
 
 inline std::ostream& operator<<(std::ostream& out, const StructuredMessage& sm) { return out << sm.msg(); }
 
+class EventMessage {
+public:
+    EventMessage(const string& msg, const std::map<string, boost::any> data = {}) : message_(msg), data_(data) {}
+
+
+    virtual ~EventMessage() {}
+
+    static constexpr const char* name = "EventMessage";
+
+    //! return a string for the log file
+    std::string msg() const { return string(name) + string(" ") + json(); }
+    void set(const std::string& key, const boost::any& value) { data_[key] = value; }
+
+private:
+    // utility function to delimate string for json, handles \" and \\ and control characters
+    string jsonify(const string& s) const;
+    string json() const;
+
+    string message_;
+    std::map<string, boost::any> data_;
+};
+
+inline std::ostream& operator<<(std::ostream& out, const EventMessage& em) { return out << em.msg(); }
+
 //! Singleton to control console logging
 //
 class ConsoleLog : public QuantLib::Singleton<ConsoleLog, std::integral_constant<bool, true>> {