fix flusher bug

alibaba · Nov 26, 2024 · 3ad7c63 · 3ad7c63
1 parent 5ffe92b
commit 3ad7c63
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/core/plugin/flusher/local_file/FlusherLocalFile.cpp b/core/plugin/flusher/local_file/FlusherLocalFile.cpp
@@ -27,6 +27,10 @@ namespace logtail {
 const string FlusherLocalFile::sName = "flusher_local_file";
 
 bool FlusherLocalFile::Init(const Json::Value& config, Json::Value& optionalGoPipeline) {
+    static uint32_t cnt = 0;
+    GenerateQueueKey(to_string(++cnt));
+    SenderQueueManager::GetInstance()->CreateQueue(mQueueKey, mPluginID, *mContext);
+
     string errorMsg;
     // FileName
     if (!GetMandatoryStringParam(config, "FileName", mFileName, errorMsg)) {
@@ -39,6 +43,8 @@ bool FlusherLocalFile::Init(const Json::Value& config, Json::Value& optionalGoPi
                            mContext->GetLogstoreName(),
                            mContext->GetRegion());
     }
+    // Pattern
+    GetMandatoryStringParam(config, "Pattern", mPattern, errorMsg);
     // MaxFileSize
     GetMandatoryUIntParam(config, "MaxFileSize", mMaxFileSize, errorMsg);
     // MaxFiles
@@ -48,8 +54,9 @@ bool FlusherLocalFile::Init(const Json::Value& config, Json::Value& optionalGoPi
     auto file_sink = std::make_shared<spdlog::sinks::rotating_file_sink_mt>(mFileName, mMaxFileSize, mMaxFiles, true);
     mFileWriter = std::make_shared<spdlog::async_logger>(
         sName, file_sink, spdlog::thread_pool(), spdlog::async_overflow_policy::block);
-    mFileWriter->set_pattern("[%Y-%m-%d %H:%M:%S.%f] %v");
+    mFileWriter->set_pattern(mPattern);
 
+    mBatcher.Init(Json::Value(), this, DefaultFlushStrategyOptions{});
     mGroupSerializer = make_unique<JsonEventGroupSerializer>(this);
     mSendCnt = GetMetricsRecordRef().CreateCounter(METRIC_PLUGIN_FLUSHER_OUT_EVENT_GROUPS_TOTAL);
     return true;
@@ -90,6 +97,7 @@ bool FlusherLocalFile::SerializeAndPush(PipelineEventGroup&& group) {
     } else {
         LOG_ERROR(sLogger, ("serialize pipeline event group error", errorMsg));
     }
+    mFileWriter->flush();
     return true;
 }
 
@@ -104,6 +112,7 @@ bool FlusherLocalFile::SerializeAndPush(BatchedEventsList&& groupList) {
             LOG_ERROR(sLogger, ("serialize pipeline event group error", errorMsg));
         }
     }
+    mFileWriter->flush();
     return true;
 }
 

diff --git a/core/plugin/flusher/local_file/FlusherLocalFile.h b/core/plugin/flusher/local_file/FlusherLocalFile.h
@@ -43,6 +43,7 @@ class FlusherLocalFile : public Flusher {
 
     std::shared_ptr<spdlog::logger> mFileWriter;
     std::string mFileName;
+    std::string mPattern = "[%Y-%m-%d %H:%M:%S.%f] %v";
     uint32_t mMaxFileSize = 1024 * 1024 * 10;
     uint32_t mMaxFiles = 10;
     Batcher<EventBatchStatus> mBatcher;