feat: rotate_file_logger supports periodic syncing (#103)

* add timer for roatet_file_logger

* add fsync and timer

* put sync task into log executor

* add doc

* rotate_file_logger support sync

- add timer to post sync_task

- add doc and example

* add readme.txt

* change head.h

* optimize code

* add release_notes

---------

Co-authored-by: hanjun <hanjun@agibot.com>

document/sphinx-cn/release_notes/v0_9_0.md

@@ -25,6 +25,7 @@
 - 现在支持 install aimrt::runtime::core；
 - 删除一些未使用的协议；
 - 支持日志自定义输出格式；
+- 支持日志定期主动落盘操作；
 - grpc 插件支持 ros2 消息以及 json 序列化格式；
 - mqtt 新增配置项以支持 ssl/tls 加密传输；
 - mqtt 插件在broker未启动时，会自动重试异步连接；

document/sphinx-cn/tutorials/cfg/log.md

@@ -121,7 +121,7 @@ aimrt:
 
 
 | 节点               | 类型         | 是否可选 | 默认值                                | 作用                                            |
-| ----------------- | ------------ | -------- | ------------------------------------- | ----------------------------- |
+| ------------------ | ------------ | -------- | ------------------------------------- | ----------------------------------------------- |
 | path               | string       | 必选     | "./log"                               | 日志文件存放目录                                |
 | filename           | string       | 必选     | "aimrt.log"                           | 日志文件基础名称                                |
 | max_file_size_m    | unsigned int | 可选     | 16                                    | 日志文件最大尺寸，单位：Mb                      |
@@ -129,6 +129,9 @@ aimrt:
 | module_filter      | string       | 可选     | "(.*)"                                | 模块过滤器                                      |
 | log_executor_name  | string       | 可选     | ""                                    | 日志执行器。默认使用主线程                      |
 | pattern            | string       | 可选     | "[%c.%f][%l][%t][%n][%g:%R:%C @%F]%v" | 日志的输出格式                                  |
+| enable_sync        | bool         | 可选     | false                                 | 是否启用定期落盘                                |
+| sync_interval_ms   | unsigned int | 可选     | 30000                                 | 定期主动落盘的时间间隔，单位：ms                |
+| sync_executor_name | string       | 条件必选 | ""                                    | 定期落盘的执行器。 `enable_sync 为 true 时必选` |
 
 
 使用注意点如下：
@@ -140,6 +143,9 @@ aimrt:
 - `module_filter`支持以正则表达式的形式，来配置哪些模块的日志可以通过本后端处理。这与模块日志等级不同，模块日志等级是全局的、先决的、影响所有日志后端的，而这里的配置只影响本后端。
 - `log_executor_name`配置了日志执行器。要求日志执行器是线程安全的，如果没有配置，则默认使用 guard 线程打印日志。
 - `pattern` 通过 `"%" + 字符` 的形式来进行格式化输出，具体可参考[console 控制台](#console-控制台日志后端)配置中的 `pattern` 字段说明。
+- `enable_sync`配置了是否启用定期落盘， 开启后数据以`sync_interval_ms`配置的时间间隔定期落盘到磁盘，以保证数据完整性， 但会降低运行性能。
+- `sync_interval_ms`配置定期落盘的时间间隔，单位：ms。 请在数据完整性和性能之间设置合适大小。
+- `sync_executor_name`配置定期落盘的执行器。用于定期落盘的执行器必须支持 timer scheduling。
 
 
 以下是一个简单的示例：

src/examples/cpp/logger/README.md

@@ -133,3 +133,38 @@
 ```
 [2024-10-31 20:35:28.378443][Thursday][Info][126632][LoggerModule][logger_module.cc] Test info log
 ```
+
+## logger rotate file with sync
+
+一个最基本的 cpp logger 示例，演示内容包括：
+-  如何使用 rotate_file 类型 Log 后端并了解其配置项；
+-  如何配置相关配置选项以开启定期落盘机制，保证数据完整性；
+
+核心代码：
+- [logger_module.cc](./module/logger_module/logger_module.cc)
+- [pkg_main.cc](./pkg/logger_pkg/pkg_main.cc)
+
+
+配置文件：
+- [examples_cpp_logger_rotate_file_with_sync_cfg.yaml](./install/linux/bin/cfg/examples_cpp_logger_rotate_file_with_sync_cfg.yaml)
+
+
+运行方式（linux）：
+- 开启 `AIMRT_BUILD_EXAMPLES` 选项编译 AimRT；
+- 直接运行 build 目录下`start_examples_cpp_logger_rotate_file.sh`脚本启动进程；
+- 键入`ctrl-c`停止进程；
+
+
+
+说明：
+- 此示例创建了一个 `LoggerModule`，会在其 `Start` 的阶段循环打印各种等级的日志；
+- 此示例会在配置文件指定的目录中 "./log" 生成的日志文件 "examples_cpp_logger_rotate_file_with_sync.log" 文件，并将日志写入其中；
+- 可以看到在该配置文件中多了如下配置：
+```yaml
+      - type: rotate_file
+        options:
+          ...
+          enable_sync: true                               # 开启定期落盘机制   
+          sync_interval_ms: 5000                          # 定期落盘间隔，单位为 ms
+          sync_executor_name: sync_timer_executor         # 指定定期落盘执行器名称，这里要和 executors 中列举的执行器列表匹配
+```

src/examples/cpp/logger/install/linux/bin/cfg/examples_cpp_logger_rotate_file_with_sync_cfg.yaml

@@ -0,0 +1,31 @@
+# Copyright (c) 2023, AgiBot Inc.
+# All rights reserved.
+
+aimrt:
+  log:
+    core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Off
+    backends:
+      - type: console
+      - type: rotate_file
+        options:
+          path: ./log
+          filename: examples_cpp_logger_rotate_file.log
+          max_file_size_m: 4
+          max_file_num: 10
+          enable_sync: true
+          sync_interval_ms: 5000
+          sync_executor_name: sync_timer_executor
+
+  executor:
+    executors:
+      - name: work_executor
+        type: simple_thread
+      - name: sync_timer_executor
+        type: asio_thread
+  module:
+    pkgs:
+      - path: ./liblogger_pkg.so
+        enable_modules: [LoggerModule]
+    modules:
+      - name: LoggerModule
+        log_lvl: TRACE

src/examples/cpp/logger/install/linux/bin/start_examples_cpp_logger_rotate_file_with_sync.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+./aimrt_main --cfg_file_path=./cfg/examples_cpp_logger_rotate_file_with_sync_cfg.yaml

src/runtime/core/logger/os.h

@@ -0,0 +1,26 @@
+// Copyright (c) 2023, AgiBot Inc.
+// All rights reserved.
+
+#pragma once
+
+#include <cstdio>
+
+#ifdef __linux__
+  #include <unistd.h>
+
+#else                   // windows
+  #include <io.h>       // for _get_osfhandle, _fileno
+  #include <windows.h>  // for FlushFileBuffers
+#endif
+
+namespace aimrt::runtime::core::logger {
+
+inline bool fsync(FILE *fp) {
+#ifdef _WIN32
+  return FlushFileBuffers(reinterpret_cast<HANDLE>(_get_osfhandle(_fileno(fp)))) != 0;
+#else
+  return ::fsync(fileno(fp)) == 0;
+#endif
+}
+
+}  // namespace aimrt::runtime::core::logger

src/runtime/core/logger/rotate_file_logger_backend.cc

@@ -2,12 +2,10 @@
 // All rights reserved.
 
 #include "core/logger/rotate_file_logger_backend.h"
-
 #include <filesystem>
 #include <map>
 #include <mutex>
 #include <regex>
-
 #include "util/exception.h"
 #include "util/string_util.h"
 
@@ -25,6 +23,9 @@ struct convert<aimrt::runtime::core::logger::RotateFileLoggerBackend::Options> {
     node["module_filter"] = rhs.module_filter;
     node["log_executor_name"] = rhs.log_executor_name;
     node["pattern"] = rhs.pattern;
+    node["enable_sync"] = rhs.enable_sync;
+    node["sync_interval_ms"] = rhs.sync_interval_ms;
+    node["sync_executor_name"] = rhs.sync_executor_name;
 
     return node;
   }
@@ -44,6 +45,12 @@ struct convert<aimrt::runtime::core::logger::RotateFileLoggerBackend::Options> {
       rhs.log_executor_name = node["log_executor_name"].as<std::string>();
     if (node["pattern"])
       rhs.pattern = node["pattern"].as<std::string>();
+    if (node["sync_interval_ms"])
+      rhs.sync_interval_ms = node["sync_interval_ms"].as<uint32_t>();
+    if (node["sync_executor_name"])
+      rhs.sync_executor_name = node["sync_executor_name"].as<std::string>();
+    if (node["enable_sync"])
+      rhs.enable_sync = node["enable_sync"].as<bool>();
 
     return true;
   }
@@ -53,6 +60,12 @@ struct convert<aimrt::runtime::core::logger::RotateFileLoggerBackend::Options> {
 namespace aimrt::runtime::core::logger {
 
 RotateFileLoggerBackend::~RotateFileLoggerBackend() {
+  if (options_.enable_sync) {
+    (void)fflush(file_);
+    (void)fclose(file_);
+    file_ = nullptr;
+    return;
+  }
   if (ofs_.is_open()) {
     ofs_.flush();
     ofs_.clear();
@@ -87,6 +100,41 @@ void RotateFileLoggerBackend::Initialize(YAML::Node options_node) {
   }
   formatter_.SetPattern(pattern_);
 
+  // if enable_sync, set sync timer
+  if (options_.enable_sync) {
+    // if enable_sync, sync_executor_name must be set
+    if (options_.sync_executor_name.empty()) {
+      throw aimrt::common::util::AimRTException("Sync executor name is empty.");
+    }
+
+    timer_executor_ = get_executor_func_(options_.sync_executor_name);
+
+    // check if timer_executor is valid
+    if (!timer_executor_) {
+      throw aimrt::common::util::AimRTException(
+          "Invalid log executor name: " + options_.sync_executor_name);
+    }
+
+    // check if sync_executor support timer schedule
+    if (!timer_executor_.SupportTimerSchedule()) {
+      throw aimrt::common::util::AimRTException("Sync executor must support timer schedule.");
+    }
+
+    // define a timer task to put sync work into log executor
+    auto timer_task = [this]() {
+      auto sync_work = [this]() {
+        (void)fflush(file_);
+        if (!logger::fsync(file_)) {
+          (void)fprintf(stderr, "sync log file:  %s failed.\n", base_file_name_.c_str());
+        }
+      };
+      log_executor_.Execute(std::move(sync_work));
+    };
+    sync_timer_ = executor::CreateTimer(timer_executor_,
+                                        std::chrono::milliseconds(options_.sync_interval_ms),
+                                        std::move(timer_task));
+  }
+
   options_node = options_;
 
   run_flag_.store(true);
@@ -102,6 +150,7 @@ void RotateFileLoggerBackend::Log(const LogDataWrapper& log_data_wrapper) noexce
 
     std::string log_data_str = formatter_.Format(log_data_wrapper);
 
+    if (!options_.enable_sync) {  // disable sync and use C++ API
       auto log_work = [this, log_data_str{std::move(log_data_str)}]() {
         if (!ofs_.is_open() || ofs_.tellp() > options_.max_file_size_m * 1024 * 1024) {
           if (!OpenNewFile()) return;
@@ -110,13 +159,29 @@ void RotateFileLoggerBackend::Log(const LogDataWrapper& log_data_wrapper) noexce
       };
 
       log_executor_.Execute(std::move(log_work));
+
+    } else {  // enable sync and use C API
+      auto log_work = [this, log_data_str{std::move(log_data_str)}]() {
+        if (!file_ || ftell(file_) > options_.max_file_size_m * 1024 * 1024) {
+          if (!OpenNewFile()) return;
+        }
+        (void)fwrite(log_data_str.data(), 1, log_data_str.size(), file_);
+        (void)fputc('\n', file_);
+        (void)fflush(file_);
+      };
+
+      log_executor_.Execute(std::move(log_work));
+    }
   } catch (const std::exception& e) {
-    fprintf(stderr, "Log get exception: %s\n", e.what());
+    (void)fprintf(stderr, "Log get exception: %s\n", e.what());
   }
 }
 
 bool RotateFileLoggerBackend::OpenNewFile() {
   bool rename_flag = false;
+
+  if (!options_.enable_sync) {  // disable sync
+    // if log file exceed max size, close it
     if (ofs_.is_open()) {
       rename_flag = (ofs_.tellp() > options_.max_file_size_m * 1024 * 1024);
       ofs_.flush();
@@ -124,17 +189,44 @@ bool RotateFileLoggerBackend::OpenNewFile() {
       ofs_.close();
     }
 
+    // rename old log file if needed
     if (rename_flag && (std::filesystem::status(base_file_name_).type() == std::filesystem::file_type::regular)) {
       std::filesystem::rename(
           base_file_name_, base_file_name_ + "_" + std::to_string(GetNextIndex()));
     }
 
+    // create and open new log file
     ofs_.open(base_file_name_, std::ios::app);
     if (!ofs_.is_open()) {
-    fprintf(stderr, "open log file %s failed.\n", base_file_name_.c_str());
+      (void)fprintf(stderr, "open log file %s failed.\n", base_file_name_.c_str());
       return false;
     }
 
+  } else {  // ensable sync
+    // if log file exceed max size, close it
+    if (file_ != NULL) {
+      (void)fseek(file_, 0, SEEK_END);
+      rename_flag = (ftell(file_) > (options_.max_file_size_m * 1024 * 1024));
+      (void)fflush(file_);
+      (void)fclose(file_);
+      file_ = nullptr;
+    }
+
+    // rename old log file if needed
+    if (rename_flag && (std::filesystem::status(base_file_name_).type() == std::filesystem::file_type::regular)) {
+      std::filesystem::rename(
+          base_file_name_, base_file_name_ + "_" + std::to_string(GetNextIndex()));
+    }
+
+    // create and open new log file
+    file_ = fopen(base_file_name_.c_str(), "a");
+    if (file_ == NULL) {
+      (void)fprintf(stderr, "open log file %s failed.\n", base_file_name_.c_str());
+      return false;
+    }
+  }
+
+  // make sure number of log files not exceed max_file_num
   CleanLogFile();
 
   return true;
@@ -214,7 +306,7 @@ bool RotateFileLoggerBackend::CheckLog(const LogDataWrapper& log_data_wrapper) {
       if_log = true;
     }
   } catch (const std::exception& e) {
-    fprintf(stderr, "Regex get exception, expr: %s, string: %s, exception info: %s\n",
+    (void)fprintf(stderr, "Regex get exception, expr: %s, string: %s, exception info: %s\n",
                   options_.module_filter.c_str(), log_data_wrapper.module_name.data(), e.what());
   }
 

src/runtime/core/logger/rotate_file_logger_backend.h

@@ -6,10 +6,11 @@
 #include <fstream>
 #include <shared_mutex>
 #include <unordered_map>
-
 #include "aimrt_module_cpp_interface/executor/executor.h"
+#include "aimrt_module_cpp_interface/executor/timer.h"
 #include "core/logger/formatter.h"
 #include "core/logger/logger_backend_base.h"
+#include "core/logger/os.h"
 #include "util/string_util.h"
 
 namespace aimrt::runtime::core::logger {
@@ -24,6 +25,9 @@ class RotateFileLoggerBackend : public LoggerBackendBase {
     std::string module_filter = "(.*)";
     std::string log_executor_name = "";
     std::string pattern;
+    bool enable_sync = false;
+    uint32_t sync_interval_ms = 30000;  // default: 30 s
+    std::string sync_executor_name = "";
   };
 
  public:
@@ -34,7 +38,14 @@ class RotateFileLoggerBackend : public LoggerBackendBase {
 
   void Initialize(YAML::Node options_node) override;
   void Start() override {}
-  void Shutdown() override { run_flag_.store(false); }
+  void Shutdown() override {
+    run_flag_.store(false);
+
+    if (sync_timer_) {
+      sync_timer_->Cancel();
+      sync_timer_->SyncWait();
+    }
+  }
 
   void RegisterGetExecutorFunc(
       const std::function<aimrt::executor::ExecutorRef(std::string_view)>& get_executor_func) {
@@ -55,6 +66,7 @@ class RotateFileLoggerBackend : public LoggerBackendBase {
   Options options_;
   std::function<aimrt::executor::ExecutorRef(std::string_view)> get_executor_func_;
   executor::ExecutorRef log_executor_;
+  executor::ExecutorRef timer_executor_;
 
   std::string base_file_name_;
   std::ofstream ofs_;
@@ -67,6 +79,10 @@ class RotateFileLoggerBackend : public LoggerBackendBase {
       module_filter_map_;
   LogFormatter formatter_;
   std::string pattern_ = "[%c.%f][%l][%t][%n][%g:%R:%C @%F]%v";
+
+  std::shared_ptr<aimrt::executor::TimerBase> sync_timer_;
+
+  FILE* file_ = nullptr;  // used for fsync()
 };
 
 }  // namespace aimrt::runtime::core::logger