feat: reload pipeline config independently

core/go_pipeline/LogtailPlugin.cpp

@@ -106,27 +106,19 @@ bool LogtailPlugin::LoadPipeline(const std::string& pipelineName,
     return false;
 }
 
-bool LogtailPlugin::UnloadPipeline(const std::string& project,
-                                   const std::string& logstore,
-                                   const std::string& pipelineName) {
+bool LogtailPlugin::UnloadPipeline(const std::string& pipelineName) {
     if (!mPluginValid) {
         LOG_ERROR(sLogger, ("UnloadPipeline", "plugin not valid"));
         return false;
     }
 
     if (mPluginValid && mUnloadPipelineFun != NULL) {
-        GoString goProject;
-        GoString goLogstore;
         GoString goConfigName;
 
-        goProject.n = project.size();
-        goProject.p = project.c_str();
-        goLogstore.n = logstore.size();
-        goLogstore.p = logstore.c_str();
         goConfigName.n = pipelineName.size();
         goConfigName.p = pipelineName.c_str();
 
-        return mUnloadPipelineFun(goProject, goLogstore, goConfigName) == 0;
+        return mUnloadPipelineFun(goConfigName) == 0;
     }
 
     return false;
@@ -139,7 +131,7 @@ void LogtailPlugin::StopAll(bool withInputFlag) {
         mStopAllFun(withInputFlag ? 1 : 0);
         auto stopAllCost = GetCurrentTimeInMilliSeconds() - stopAllStart;
         LOG_INFO(sLogger, ("Go pipelines stop all", "succeeded")("cost", ToString(stopAllCost) + "ms"));
-        if (stopAllCost >= 60 * 1000) {
+        if (stopAllCost >= 10 * 1000) {
             LogtailAlarm::GetInstance()->SendAlarm(HOLD_ON_TOO_SLOW_ALARM,
                                                    "Stopping all Go pipelines took " + ToString(stopAllCost) + "ms");
         }
@@ -156,13 +148,21 @@ void LogtailPlugin::Stop(const std::string& configName, bool removedFlag) {
         mStopFun(goConfigName, removedFlag ? 1 : 0);
         auto stopCost = GetCurrentTimeInMilliSeconds() - stopStart;
         LOG_INFO(sLogger, ("Go pipelines stop", "succeeded")("config", configName)("cost", ToString(stopCost) + "ms"));
-        if (stopCost >= 30 * 1000) {
+        if (stopCost >= 10 * 1000) {
             LogtailAlarm::GetInstance()->SendAlarm(
                 HOLD_ON_TOO_SLOW_ALARM, "Stopping Go pipeline " + configName + " took " + ToString(stopCost) + "ms");
         }
     }
 }
 
+void LogtailPlugin::StopBuiltIn() {
+    if (mPluginValid && mStopFun != NULL) {
+        LOG_INFO(sLogger, ("Go pipelines stop built-in", "starts"));
+        mStopBuiltInFun();
+        LOG_INFO(sLogger, ("Go pipelines stop built-in", "succeeded"));
+    }
+}
+
 void LogtailPlugin::Start(const std::string& configName) {
     if (mPluginValid && mStartFun != NULL) {
         LOG_INFO(sLogger, ("Go pipelines start", "starts")("config name", configName));
@@ -397,6 +397,12 @@ bool LogtailPlugin::LoadPluginBase() {
             LOG_ERROR(sLogger, ("load Stop error, Message", error));
             return mPluginValid;
         }
+        // 停止内置功能
+        mStopBuiltInFun = (StopBuiltInFun)loader.LoadMethod("StopBuiltIn", error);
+        if (!error.empty()) {
+            LOG_ERROR(sLogger, ("load StopBuiltIn error, Message", error));
+            return mPluginValid;
+        }
         // 插件恢复
         mStartFun = (StartFun)loader.LoadMethod("Start", error);
         if (!error.empty()) {

core/go_pipeline/LogtailPlugin.h

@@ -133,9 +133,10 @@ struct K8sContainerMeta {
 // Methods export by plugin.
 typedef GoInt (*LoadGlobalConfigFun)(GoString);
 typedef GoInt (*LoadPipelineFun)(GoString p, GoString l, GoString c, GoInt64 k, GoString p2);
-typedef GoInt (*UnloadPipelineFun)(GoString p, GoString l, GoString c);
+typedef GoInt (*UnloadPipelineFun)(GoString c);
 typedef void (*StopAllFun)(GoInt);
 typedef void (*StopFun)(GoString, GoInt);
+typedef void (*StopBuiltInFun)();
 typedef void (*StartFun)(GoString);
 typedef GoInt (*InitPluginBaseFun)();
 typedef GoInt (*InitPluginBaseV2Fun)(GoString cfg);
@@ -212,9 +213,10 @@ class LogtailPlugin {
                       const std::string& logstore = "",
                       const std::string& region = "",
                       logtail::QueueKey logstoreKey = 0);
-    bool UnloadPipeline(const std::string& project, const std::string& logstore, const std::string& pipelineName);
+    bool UnloadPipeline(const std::string& pipelineName);
     void StopAll(bool withInputFlag);
     void Stop(const std::string& configName, bool removingFlag);
+    void StopBuiltIn();
     void Start(const std::string& configName);
 
     bool IsPluginOpened() { return mPluginValid; }
@@ -264,6 +266,7 @@ class LogtailPlugin {
     UnloadPipelineFun mUnloadPipelineFun;
     StopAllFun mStopAllFun;
     StopFun mStopFun;
+    StopBuiltInFun mStopBuiltInFun;
     StartFun mStartFun;
     volatile bool mPluginValid;
     logtail::FlusherSLS mPluginAlarmConfig;

core/pipeline/Pipeline.cpp

@@ -415,7 +415,7 @@ void Pipeline::Stop(bool isRemoving) {
     }
 
     ProcessQueueManager::GetInstance()->DisablePop(mName, isRemoving);
-    WaitAllInProcessFinish();
+    WaitAllItemsInProcessFinished();
 
     if (!isRemoving) {
         FlushBatch();
@@ -514,8 +514,7 @@ bool Pipeline::LoadGoPipelines() const {
                                                    mContext.GetLogstoreName(),
                                                    mContext.GetRegion());
             if (!mGoPipelineWithoutInput.isNull()) {
-                LogtailPlugin::GetInstance()->UnloadPipeline(
-                    mContext.GetProjectName(), mContext.GetLogstoreName(), GetConfigNameOfGoPipelineWithoutInput());
+                LogtailPlugin::GetInstance()->UnloadPipeline(GetConfigNameOfGoPipelineWithoutInput());
             }
             return false;
         }
@@ -539,13 +538,13 @@ PluginInstance::PluginMeta Pipeline::GenNextPluginMeta(bool lastOne) {
         std::to_string(mPluginID.load()), std::to_string(mPluginID.load()), std::to_string(childNodeID));
 }
 
-void Pipeline::WaitAllInProcessFinish() {
+void Pipeline::WaitAllItemsInProcessFinished() {
     uint64_t startTime = GetCurrentTimeInMilliSeconds();
     bool alarmOnce = false;
     while (mInProcessCnt.load() != 0) {
-        usleep(1000 * 10); // 10ms
+        this_thread::sleep_for(chrono::milliseconds(100)); // 100ms
         uint64_t duration = GetCurrentTimeInMilliSeconds() - startTime;
-        if (!alarmOnce && duration > 1000) { // 1s
+        if (!alarmOnce && duration > 10000) { // 10s
             LOG_ERROR(sLogger, ("pipeline stop", "too slow")("config", mName)("cost", duration));
             LogtailAlarm::GetInstance()->SendAlarm(CONFIG_UPDATE_ALARM,
                                                    string("pipeline stop too slow, config: ") + mName

core/pipeline/Pipeline.h

@@ -89,7 +89,7 @@ class Pipeline {
                                Json::Value& dst);
     void CopyNativeGlobalParamToGoPipeline(Json::Value& root);
     bool ShouldAddPluginToGoPipelineWithInput() const { return mInputs.empty() && mProcessorLine.empty(); }
-    void WaitAllInProcessFinish();
+    void WaitAllItemsInProcessFinished();
 
     std::string mName;
     std::vector<std::unique_ptr<InputInstance>> mInputs;
@@ -110,7 +110,6 @@ class Pipeline {
     CounterPtr mProcessorsInGroupsCnt;
     CounterPtr mProcessorsInGroupDataSizeBytes;
     CounterPtr mProcessorsTotalDelayMs;
-    CounterPtr mLoadDelayMs;
 
 #ifdef APSARA_UNIT_TEST_MAIN
     friend class PipelineMock;

core/pipeline/PipelineManager.cpp

@@ -96,10 +96,9 @@ void logtail::PipelineManager::UpdatePipelines(PipelineConfigDiff& diff) {
         ConfigFeedbackReceiver::GetInstance().FeedbackPipelineConfigStatus(name, ConfigFeedbackStatus::DELETED);
     }
     for (auto& config : diff.mModified) {
-        auto iter = mPipelineNameEntityMap.find(config.mName);
         auto p = BuildPipeline(std::move(config)); // auto reuse old pipeline's process queue and sender queue
         if (!p) {
-            LOG_WARNING(sLogger,
+                        LOG_WARNING(sLogger,
                         ("failed to build pipeline for existing config",
                          "keep current pipeline running")("config", config.mName));
             LogtailAlarm::GetInstance()->SendAlarm(
@@ -115,6 +114,7 @@ void logtail::PipelineManager::UpdatePipelines(PipelineConfigDiff& diff) {
         LOG_INFO(sLogger,
                  ("pipeline building for existing config succeeded",
                   "stop the old pipeline and start the new one")("config", config.mName));
+        auto iter = mPipelineNameEntityMap.find(config.mName);
         iter->second->Stop(false);
         DecreasePluginUsageCnt(iter->second->GetPluginStatistics());
 
@@ -126,7 +126,7 @@ void logtail::PipelineManager::UpdatePipelines(PipelineConfigDiff& diff) {
     for (auto& config : diff.mAdded) {
         auto p = BuildPipeline(std::move(config));
         if (!p) {
-            LOG_WARNING(sLogger,
+                        LOG_WARNING(sLogger,
                         ("failed to build pipeline for new config", "skip current object")("config", config.mName));
             LogtailAlarm::GetInstance()->SendAlarm(
                 CATEGORY_CONFIG_ALARM,
@@ -231,6 +231,7 @@ void PipelineManager::StopAllPipelines() {
     FileServer::GetInstance()->Stop();
 
     LogtailPlugin::GetInstance()->StopAll(true);
+    LogtailPlugin::GetInstance()->StopBuiltIn();
 
     bool logProcessFlushFlag = false;
     for (int i = 0; !logProcessFlushFlag && i < 500; ++i) {

core/pipeline/plugin/interface/Flusher.cpp

@@ -36,15 +36,13 @@ bool Flusher::Stop(bool isPipelineRemoving) {
 }
 
 void Flusher::SetPipelineForItemsWhenStop() {
-    vector<SenderQueueItem*> items;
-    SenderQueueManager::GetInstance()->GetAllAvailableItems(mQueueKey, items, false);
     if (HasContext()) {
         auto pipeline = PipelineManager::GetInstance()->FindConfigByName(mContext->GetConfigName());
-        for (auto item : items) {
-            item->mPipeline = pipeline;
+        if (!pipeline) {
+            LOG_ERROR(sLogger, ("failed to get pipeline context", "context not found")("action", "not set pipeline"));
+            return;
         }
-    } else {
-        LOG_ERROR(sLogger, ("failed to get pipeline context", "context not found")("action", "discard data"));
+        SenderQueueManager::GetInstance()->SetPipelineForItems(mQueueKey, pipeline);
     }
 }
 

core/pipeline/plugin/interface/Flusher.h

@@ -58,6 +58,7 @@ class Flusher : public Plugin {
 #ifdef APSARA_UNIT_TEST_MAIN
     friend class FlusherInstanceUnittest;
     friend class FlusherRunnerUnittest;
+    friend class FlusherUnittest;
 #endif
 };
 

core/pipeline/queue/BoundedProcessQueue.cpp

@@ -68,8 +68,7 @@ bool BoundedProcessQueue::Pop(unique_ptr<ProcessQueueItem>& item) {
     return true;
 }
 
-void BoundedProcessQueue::SetPipelineForItems(const std::string& name) const {
-    auto p = PipelineManager::GetInstance()->FindConfigByName(name);
+void BoundedProcessQueue::SetPipelineForItems(std::shared_ptr<Pipeline>& p) const {
     for (auto& item : mQueue) {
         if (!item->mPipeline) {
             item->mPipeline = p;

core/pipeline/queue/BoundedProcessQueue.h

@@ -31,11 +31,12 @@ namespace logtail {
 class BoundedProcessQueue : public BoundedQueueInterface<std::unique_ptr<ProcessQueueItem>>,
                             public ProcessQueueInterface {
 public:
-    BoundedProcessQueue(size_t cap, size_t low, size_t high, int64_t key, uint32_t priority, const PipelineContext& ctx);
+    BoundedProcessQueue(
+        size_t cap, size_t low, size_t high, int64_t key, uint32_t priority, const PipelineContext& ctx);
 
     bool Push(std::unique_ptr<ProcessQueueItem>&& item) override;
     bool Pop(std::unique_ptr<ProcessQueueItem>& item) override;
-    void SetPipelineForItems(const std::string& name) const override;
+    void SetPipelineForItems(std::shared_ptr<Pipeline>& p) const override;
 
     void SetUpStreamFeedbacks(std::vector<FeedbackInterface*>&& feedbacks);
 

core/pipeline/queue/BoundedSenderQueueInterface.h

@@ -47,6 +47,7 @@ class BoundedSenderQueueInterface : public BoundedQueueInterface<std::unique_ptr
 
     void SetRateLimiter(uint32_t maxRate);
     void SetConcurrencyLimiters(std::vector<std::shared_ptr<ConcurrencyLimiter>>&& limiters);
+    virtual void SetPipelineForItems(std::shared_ptr<Pipeline>& p) const = 0;
 
 #ifdef APSARA_UNIT_TEST_MAIN
     std::optional<RateLimiter>& GetRateLimiter() { return mRateLimiter; }

core/pipeline/queue/CircularProcessQueue.cpp

@@ -73,8 +73,7 @@ bool CircularProcessQueue::Pop(unique_ptr<ProcessQueueItem>& item) {
     return true;
 }
 
-void CircularProcessQueue::SetPipelineForItems(const std::string& name) const {
-    auto p = PipelineManager::GetInstance()->FindConfigByName(name);
+void CircularProcessQueue::SetPipelineForItems(std::shared_ptr<Pipeline>& p) const {
     for (auto& item : mQueue) {
         if (!item->mPipeline) {
             item->mPipeline = p;

core/pipeline/queue/CircularProcessQueue.h

@@ -33,7 +33,7 @@ class CircularProcessQueue : virtual public QueueInterface<std::unique_ptr<Proce
 
     bool Push(std::unique_ptr<ProcessQueueItem>&& item) override;
     bool Pop(std::unique_ptr<ProcessQueueItem>& item) override;
-    void SetPipelineForItems(const std::string& name) const override;
+    void SetPipelineForItems(std::shared_ptr<Pipeline>& p) const override;
 
     void Reset(size_t cap);
 

core/pipeline/queue/ExactlyOnceQueueManager.cpp

@@ -16,11 +16,11 @@
 
 #include "common/Flags.h"
 #include "common/TimeUtil.h"
-#include "plugin/input/InputFeedbackInterfaceRegistry.h"
-#include "plugin/input/InputFile.h"
 #include "logger/Logger.h"
 #include "pipeline/queue/ProcessQueueManager.h"
 #include "pipeline/queue/QueueKeyManager.h"
+#include "plugin/input/InputFeedbackInterfaceRegistry.h"
+#include "plugin/input/InputFile.h"
 
 DEFINE_FLAG_INT32(logtail_queue_gc_threshold_sec, "2min", 2 * 60);
 DEFINE_FLAG_INT64(logtail_queue_max_used_time_per_round_in_msec, "500ms", 500);
@@ -154,7 +154,10 @@ void ExactlyOnceQueueManager::DisablePopProcessQueue(const string& configName, b
         if (iter.second->GetConfigName() == configName) {
             iter.second->DisablePop();
             if (!isPipelineRemoving) {
-                iter.second->SetPipelineForItems(configName);
+                auto p = PipelineManager::GetInstance()->FindConfigByName(configName);
+                if (p) {
+                    iter.second->SetPipelineForItems(p);
+                }
             }
         }
     }
@@ -276,6 +279,14 @@ uint32_t ExactlyOnceQueueManager::GetProcessQueueCnt() const {
     return mProcessQueues.size();
 }
 
+void ExactlyOnceQueueManager::SetPipelineForSenderItems(QueueKey key, std::shared_ptr<Pipeline>& p) {
+    lock_guard<mutex> lock(mSenderQueueMux);
+    auto iter = mSenderQueues.find(key);
+    if (iter != mSenderQueues.end()) {
+        iter->second.SetPipelineForItems(p);
+    }
+}
+
 #ifdef APSARA_UNIT_TEST_MAIN
 void ExactlyOnceQueueManager::Clear() {
     {

core/pipeline/queue/ExactlyOnceQueueManager.h

@@ -67,6 +67,7 @@ class ExactlyOnceQueueManager {
     void GetAllAvailableSenderQueueItems(std::vector<SenderQueueItem*>& item, bool withLimits = true);
     bool RemoveSenderQueueItem(QueueKey key, SenderQueueItem* item);
     bool IsAllSenderQueueEmpty() const;
+    void SetPipelineForSenderItems(QueueKey key, std::shared_ptr<Pipeline>& p);
 
     void ClearTimeoutQueues();
 

core/pipeline/queue/ExactlyOnceSenderQueue.cpp

@@ -153,4 +153,12 @@ void ExactlyOnceSenderQueue::Reset(const vector<RangeCheckpointPtr>& checkpoints
     mRangeCheckpoints = checkpoints;
 }
 
+void ExactlyOnceSenderQueue::SetPipelineForItems(std::shared_ptr<Pipeline>& p) const {
+    for (size_t index = 0; index < mCapacity; ++index) {
+        if (!mQueue[index]) {
+            mQueue[index]->mPipeline = p;
+        }
+    }
+}
+
 } // namespace logtail

core/pipeline/queue/ExactlyOnceSenderQueue.h

@@ -30,11 +30,14 @@ namespace logtail {
 // not thread-safe, should be protected explicitly by queue manager
 class ExactlyOnceSenderQueue : public BoundedSenderQueueInterface {
 public:
-    ExactlyOnceSenderQueue(const std::vector<RangeCheckpointPtr>& checkpoints, QueueKey key, const PipelineContext& ctx);
+    ExactlyOnceSenderQueue(const std::vector<RangeCheckpointPtr>& checkpoints,
+                           QueueKey key,
+                           const PipelineContext& ctx);
 
     bool Push(std::unique_ptr<SenderQueueItem>&& item) override;
     bool Remove(SenderQueueItem* item) override;
     void GetAllAvailableItems(std::vector<SenderQueueItem*>& items, bool withLimits = true) override;
+    void SetPipelineForItems(std::shared_ptr<Pipeline>& p) const override;
 
     void Reset(const std::vector<RangeCheckpointPtr>& checkpoints);
 

core/pipeline/queue/ProcessQueueInterface.h

@@ -45,7 +45,7 @@ class ProcessQueueInterface : virtual public QueueInterface<std::unique_ptr<Proc
     void DisablePop() { mValidToPop = false; }
     void EnablePop() { mValidToPop = true; }
 
-    virtual void SetPipelineForItems(const std::string& name) const = 0;
+    virtual void SetPipelineForItems(std::shared_ptr<Pipeline>& p) const = 0;
 
     void Reset() { mDownStreamQueues.clear(); }
 

core/pipeline/queue/ProcessQueueManager.cpp

@@ -237,7 +237,10 @@ void ProcessQueueManager::DisablePop(const string& configName, bool isPipelineRe
         if (iter != mQueues.end()) {
             (*iter->second.first)->DisablePop();
             if (!isPipelineRemoving) {
-                (*iter->second.first)->SetPipelineForItems(configName);
+                auto p = PipelineManager::GetInstance()->FindConfigByName(configName);
+                if (p) {
+                    (*iter->second.first)->SetPipelineForItems(p);
+                }
             }
         }
     } else {

core/pipeline/queue/SenderQueue.cpp

@@ -142,4 +142,12 @@ void SenderQueue::GetAllAvailableItems(vector<SenderQueueItem*>& items, bool wit
     }
 }
 
+void SenderQueue::SetPipelineForItems(std::shared_ptr<Pipeline>& p) const {
+    for (auto& item : mQueue) {
+        if (!item->mPipeline) {
+            item->mPipeline = p;
+        }
+    }
+}
+
 } // namespace logtail

core/pipeline/queue/SenderQueue.h

@@ -36,6 +36,7 @@ class SenderQueue : public BoundedSenderQueueInterface {
     bool Push(std::unique_ptr<SenderQueueItem>&& item) override;
     bool Remove(SenderQueueItem* item) override;
     void GetAllAvailableItems(std::vector<SenderQueueItem*>& items, bool withLimits = true) override;
+    void SetPipelineForItems(std::shared_ptr<Pipeline>& p) const override;
 
 private:
     size_t Size() const override { return mSize; }