[filter] Support shared mode and output file rotating in tee() filter

Author: pajama-coder

src/api/logging.cpp

@@ -118,7 +118,7 @@ bool Logger::tail(const std::string &name, Data &buffer) {
 }
 
 void Logger::close_all() {
-  FileTarget::close_all_writers();
+
 }
 
 Logger::Logger(pjs::Str *name)
@@ -305,69 +305,24 @@ void Logger::StdoutTarget::write(const Data &msg) {
 // Logger::FileTarget
 //
 
-void Logger::FileTarget::close_all_writers() {
-  s_all_writers.clear();
-}
-
-Logger::FileTarget::FileTarget(pjs::Str *filename)
+Logger::FileTarget::FileTarget(pjs::Str *filename, const Options &options)
   : m_filename(pjs::Str::make(fs::abs_path(filename->str())))
-{
-}
-
-void Logger::FileTarget::write(const Data &msg) {
-  auto name = m_filename->data()->retain();
-  auto sd = SharedData::make(msg)->retain();
-  Net::main().post(
-    [=]() {
-      Writer *writer = nullptr;
-      const auto &filename = name->str();
-      auto i = s_all_writers.find(filename);
-      if (i != s_all_writers.end()) {
-        writer = i->second.get();
-      } else {
-        writer = new Writer(filename);
-        s_all_writers[filename].reset(writer);
-      }
-      InputContext ic;
-      Data data;
-      sd->to_data(data);
-      writer->write(data);
-      name->release();
-      sd->release();
-    }
-  );
-}
-
-//
-// Logger::FileTarget::Writer
-//
-
-std::map<std::string, std::unique_ptr<Logger::FileTarget::Writer>> Logger::FileTarget::s_all_writers;
-
-Logger::FileTarget::Writer::Writer(const std::string &filename)
-  : m_module(new Module)
+  , m_options(options)
+  , m_module(new Module)
 {
   PipelineLayout *ppl = PipelineLayout::make();
-  Tee::Options options;
-  options.append = true;
   ppl->append(new Tee(filename, options));
-
   m_pipeline_layout = ppl;
   m_pipeline = Pipeline::make(ppl, Context::make());
 }
 
-void Logger::FileTarget::Writer::write(const Data &msg) {
+void Logger::FileTarget::write(const Data &msg) {
   Data *buf = Data::make();
   s_dp.push(buf, &msg);
   s_dp.push(buf, '\n');
   m_pipeline->input()->input(buf);
 }
 
-void Logger::FileTarget::Writer::shutdown() {
-  m_module->shutdown();
-  m_pipeline = nullptr;
-}
-
 //
 // Logger::SyslogTarget
 //
@@ -605,8 +560,9 @@ template<> void ClassDef<Logger>::init() {
 
   method("toFile", [](Context &ctx, Object *obj, Value &ret) {
     pjs::Str *filename;
-    if (!ctx.arguments(1, &filename)) return;
-    obj->as<Logger>()->add_target(new Logger::FileTarget(filename));
+    pjs::Object *options = nullptr;
+    if (!ctx.arguments(1, &filename, &options)) return;
+    obj->as<Logger>()->add_target(new Logger::FileTarget(filename, options));
     ret.set(obj);
   });
 

src/api/logging.hpp

@@ -31,6 +31,7 @@
 #include "module.hpp"
 #include "fstream.hpp"
 #include "filters/pack.hpp"
+#include "filters/tee.hpp"
 #include "filters/tls.hpp"
 
 #include <atomic>
@@ -97,9 +98,12 @@ class Logger : public pjs::ObjectTemplate<Logger> {
 
   class FileTarget : public Target {
   public:
-    static void close_all_writers();
+    struct Options : public Tee::Options {
+      Options() { shared = true; append = true; }
+      Options(pjs::Object *options) : Tee::Options(options) { shared = true; append = true; }
+    };
 
-    FileTarget(pjs::Str *filename);
+    FileTarget(pjs::Str *filename, const Options &options = Options());
 
   private:
     virtual void write(const Data &msg) override;
@@ -116,24 +120,11 @@ class Logger : public pjs::ObjectTemplate<Logger> {
       }
     };
 
-    //
-    // Logger::FileTarget::Writer
-    //
-
-    class Writer {
-    public:
-      Writer(const std::string &filename);
-      void write(const Data &msg);
-      void shutdown();
-    private:
-      pjs::Ref<Module> m_module;
-      pjs::Ref<PipelineLayout> m_pipeline_layout;
-      pjs::Ref<Pipeline> m_pipeline;
-    };
-
     pjs::Ref<pjs::Str> m_filename;
-
-    static std::map<std::string, std::unique_ptr<Writer>> s_all_writers;
+    Options m_options;
+    pjs::Ref<Module> m_module;
+    pjs::Ref<PipelineLayout> m_pipeline_layout;
+    pjs::Ref<Pipeline> m_pipeline;
   };
 
   //

src/filters/tee.cpp

@@ -24,7 +24,11 @@
  */
 
 #include "tee.hpp"
-#include "log.hpp"
+#include "fs.hpp"
+#include "input.hpp"
+#include "utils.hpp"
+
+#include <algorithm>
 
 namespace pipy {
 
@@ -33,9 +37,21 @@ namespace pipy {
 //
 
 Tee::Options::Options(pjs::Object *options) {
+  Value(options, "shared")
+    .get(shared)
+    .check_nullable();
   Value(options, "append")
     .get(append)
     .check_nullable();
+  Value(options, "maxFileSize")
+    .get(max_file_size)
+    .check_nullable();
+  Value(options, "maxFileCount")
+    .get(max_file_count)
+    .check_nullable();
+  Value(options, "rotateInterval")
+    .get(rotate_interval)
+    .check_nullable();
 }
 
 //
@@ -73,6 +89,7 @@ void Tee::reset() {
     m_file->close();
     m_file = nullptr;
   }
+  m_target = nullptr;
   m_resolved_filename = nullptr;
 }
 
@@ -84,12 +101,18 @@ void Tee::process(Event *evt) {
       auto *s = filename.to_string();
       m_resolved_filename = s;
       s->release();
-      m_file = File::make(m_resolved_filename->str());
-      m_file->open_write(m_options.append);
+      if (m_options.shared) {
+        m_target = get_target(m_resolved_filename->str(), m_options);
+      } else {
+        m_file = File::make(m_resolved_filename->str());
+        m_file->open_write(m_options.append);
+      }
     }
 
     if (m_file) {
       m_file->write(*data);
+    } else if (m_target) {
+      m_target->write(*data);
     }
 
   } else if (evt->is<StreamEnd>()) {
@@ -102,4 +125,109 @@ void Tee::process(Event *evt) {
   output(evt);
 }
 
+std::map<std::string, pjs::Ref<Tee::Target>> Tee::s_targets;
+std::mutex Tee::s_targets_mutex;
+
+auto Tee::get_target(const std::string &filename, const Options &options) -> Target* {
+  std::lock_guard<std::mutex> lock(s_targets_mutex);
+  auto path = filename == "-" ? filename : fs::abs_path(filename);
+  auto i = s_targets.find(path);
+  if (i != s_targets.end()) return i->second;
+  return s_targets[path] = new Target(path, options);
+}
+
+Tee::Target::Target(const std::string &filename, const Tee::Options &options)
+  : m_filename(filename)
+  , m_options(options)
+{
+}
+
+Tee::Target::~Target() {
+
+}
+
+void Tee::Target::write(const Data &data) {
+  if (Net::main().is_running()) {
+    auto sd = SharedData::make(data)->retain();
+    Net::main().post([=]() {
+      Net::main().post(
+        [=]() {
+          Data data;
+          sd->to_data(data);
+          write_async(data);
+          sd->release();
+        }
+      );
+    });
+  } else {
+    write_async(data);
+  }
+}
+
+void Tee::Target::write_async(const Data &data) {
+  if (m_file && m_written_size > 0 && m_filename != "-" && (
+    (m_options.max_file_size > 0 && m_written_size + data.size() > m_options.max_file_size) ||
+    (m_options.rotate_interval > 0 && utils::now() - m_file_time > m_options.rotate_interval * 1000)
+  )) {
+    m_file->close();
+    m_file = nullptr;
+
+    auto sec = std::floor(m_file_time / 1000);
+    auto t = std::time_t(sec);
+    std::tm tm;
+    localtime_r(&t, &tm);
+
+    char str[100];
+    auto len = std::strftime(str, sizeof(str), "%Y-%m-%d-%H-%M-%S-", &tm);
+    auto date_filename = utils::path_join(
+      utils::path_dirname(m_filename),
+      std::string(str, len) + utils::path_basename(m_filename)
+    );
+
+    fs::rename(m_filename, date_filename);
+
+    if (m_options.max_file_count > 0) {
+      auto dirname = utils::path_dirname(m_filename);
+      auto basename = utils::path_basename(m_filename);
+      std::list<std::string> all;
+      fs::read_dir(dirname, all);
+      std::vector<std::string> names;
+      for (const auto &name : all) {
+        if (utils::ends_with(name, basename)) {
+          names.push_back(name);
+        }
+      }
+      if (names.size() > m_options.max_file_count) {
+        std::sort(
+          names.begin(), names.end(),
+          [](const std::string &a, const std::string &b) -> bool { return a > b; }
+        );
+        while (names.size() > m_options.max_file_count) {
+          auto name = names.back();
+          names.pop_back();
+          fs::unlink(utils::path_join(dirname, name));
+        }
+      }
+    }
+  }
+
+  if (!m_file) {
+    m_file_time = utils::now();
+    m_written_size = 0;
+    if (m_filename != "-") {
+      fs::Stat st;
+      if (fs::stat(m_filename, st) && st.is_file()) {
+        m_file_time = st.ctime * 1000;
+        m_written_size = st.size;
+      }
+    }
+    m_file = File::make(m_filename);
+    m_file->open_write(m_options.append);
+  }
+
+  InputContext ic;
+  m_file->write(data);
+  m_written_size += data.size();
+}
+
 } // namespace pipy

src/filters/tee.hpp

@@ -32,6 +32,9 @@
 #include "options.hpp"
 #include "fstream.hpp"
 
+#include <atomic>
+#include <string>
+
 namespace pipy {
 
 //
@@ -41,11 +44,32 @@ namespace pipy {
 class Tee : public Filter {
 public:
   struct Options : public pipy::Options {
+    bool shared = false;
     bool append = false;
+    int max_file_size = 0;
+    int max_file_count = 0;
+    double rotate_interval = 0;
     Options() {}
     Options(pjs::Object *options);
   };
 
+  class Target : public pjs::RefCountMT<Target> {
+  public:
+    Target(const std::string &filename, const Options &options);
+    ~Target();
+
+    void write(const Data &data);
+
+  private:
+    const std::string m_filename;
+    Options m_options;
+    pjs::Ref<File> m_file;
+    int m_written_size = 0;
+    double m_file_time = 0;
+
+    void write_async(const Data &data);
+  };
+
   Tee(const pjs::Value &filename, const Options &options);
 
 private:
@@ -61,6 +85,12 @@ class Tee : public Filter {
   pjs::Value m_filename;
   pjs::Ref<pjs::Str> m_resolved_filename;
   pjs::Ref<File> m_file;
+  pjs::Ref<Target> m_target;
+
+  static std::map<std::string, pjs::Ref<Target>> s_targets;
+  static std::mutex s_targets_mutex;
+
+  static auto get_target(const std::string &filename, const Options &options) -> Target*;
 };
 
 } // namespace pipy