fix

core/go_pipeline/LogtailPlugin.cpp

@@ -47,7 +47,7 @@ LogtailPlugin::LogtailPlugin() {
     mPluginBasePtr = NULL;
     mLoadPipelineFun = NULL;
     mUnloadPipelineFun = NULL;
-    mStopAllFun = NULL;
+    mStopAllPipelinesFun = NULL;
     mStopFun = NULL;
     mStartFun = NULL;
     mLoadGlobalConfigFun = NULL;
@@ -124,11 +124,11 @@ bool LogtailPlugin::UnloadPipeline(const std::string& pipelineName) {
     return false;
 }
 
-void LogtailPlugin::StopAll(bool withInputFlag) {
-    if (mPluginValid && mStopAllFun != NULL) {
+void LogtailPlugin::StopAllPipelines(bool withInputFlag) {
+    if (mPluginValid && mStopAllPipelinesFun != NULL) {
         LOG_INFO(sLogger, ("Go pipelines stop all", "starts"));
         auto stopAllStart = GetCurrentTimeInMilliSeconds();
-        mStopAllFun(withInputFlag ? 1 : 0);
+        mStopAllPipelinesFun(withInputFlag ? 1 : 0);
         auto stopAllCost = GetCurrentTimeInMilliSeconds() - stopAllStart;
         LOG_INFO(sLogger, ("Go pipelines stop all", "succeeded")("cost", ToString(stopAllCost) + "ms"));
         if (stopAllCost >= 10 * 1000) {
@@ -386,9 +386,9 @@ bool LogtailPlugin::LoadPluginBase() {
             return mPluginValid;
         }
         // 停止所有插件
-        mStopAllFun = (StopAllFun)loader.LoadMethod("StopAll", error);
+        mStopAllPipelinesFun = (StopAllPipelinesFun)loader.LoadMethod("StopAllPipelines", error);
         if (!error.empty()) {
-            LOG_ERROR(sLogger, ("load StopAll error, Message", error));
+            LOG_ERROR(sLogger, ("load StopAllPipelines error, Message", error));
             return mPluginValid;
         }
         // 停止单个插件

core/go_pipeline/LogtailPlugin.h

@@ -134,7 +134,7 @@ struct K8sContainerMeta {
 typedef GoInt (*LoadGlobalConfigFun)(GoString);
 typedef GoInt (*LoadPipelineFun)(GoString p, GoString l, GoString c, GoInt64 k, GoString p2);
 typedef GoInt (*UnloadPipelineFun)(GoString c);
-typedef void (*StopAllFun)(GoInt);
+typedef void (*StopAllPipelinesFun)(GoInt);
 typedef void (*StopFun)(GoString, GoInt);
 typedef void (*StopBuiltInModulesFun)();
 typedef void (*StartFun)(GoString);
@@ -214,7 +214,7 @@ class LogtailPlugin {
                       const std::string& region = "",
                       logtail::QueueKey logstoreKey = 0);
     bool UnloadPipeline(const std::string& pipelineName);
-    void StopAll(bool withInputFlag);
+    void StopAllPipelines(bool withInputFlag);
     void Stop(const std::string& configName, bool removingFlag);
     void StopBuiltInModules();
     void Start(const std::string& configName);
@@ -264,7 +264,7 @@ class LogtailPlugin {
     LoadGlobalConfigFun mLoadGlobalConfigFun;
     LoadPipelineFun mLoadPipelineFun;
     UnloadPipelineFun mUnloadPipelineFun;
-    StopAllFun mStopAllFun;
+    StopAllPipelinesFun mStopAllPipelinesFun;
     StopFun mStopFun;
     StopBuiltInModulesFun mStopBuiltInModulesFun;
     StartFun mStartFun;

core/pipeline/PipelineManager.cpp

@@ -148,6 +148,12 @@ void logtail::PipelineManager::UpdatePipelines(PipelineConfigDiff& diff) {
     }
 #endif
 #endif
+
+    for (auto& item : mInputRunners) {
+        if (!item->HasRegisteredPlugins()) {
+            item->Stop();
+        }
+    }
 }
 
 shared_ptr<Pipeline> PipelineManager::FindConfigByName(const string& configName) const {
@@ -180,19 +186,17 @@ string PipelineManager::GetPluginStatistics() const {
 void PipelineManager::StopAllPipelines() {
     LOG_INFO(sLogger, ("stop all pipelines", "starts"));
     for (auto& item : mInputRunners) {
-        if (item->HasRegisteredPlugins()) {
-            item->Stop();
-        }
+        item->Stop();
     }
     FileServer::GetInstance()->Stop();
 
-    LogtailPlugin::GetInstance()->StopAll(true);
+    LogtailPlugin::GetInstance()->StopAllPipelines(true);
 
     ProcessorRunner::GetInstance()->Stop();
 
     FlushAllBatch();
 
-    LogtailPlugin::GetInstance()->StopAll(false);
+    LogtailPlugin::GetInstance()->StopAllPipelines(false);
 
     // TODO: make it common
     FlusherSLS::RecycleResourceIfNotUsed();

plugin_main/plugin_export.go

@@ -186,16 +186,20 @@ func ProcessLogGroup(configName string, logBytes []byte, packID string) int {
 	return config.ProcessLogGroup(logBytes, util.StringDeepCopy(packID))
 }
 
-//export StopAll
-func StopAll(withInputFlag int) {
-	logger.Info(context.Background(), "Stop all", "start")
-	err := pluginmanager.StopAll(withInputFlag != 0)
+//export StopAllPipelines
+func StopAllPipelines(withInputFlag int) {
+	logger.Info(context.Background(), "Stop all", "start", "with input", withInputFlag)
+	err := pluginmanager.StopAllPipelines(withInputFlag != 0)
 	if err != nil {
 		logger.Error(context.Background(), "PLUGIN_ALARM", "stop all error", err)
 	}
-	logger.Info(context.Background(), "Stop all", "success")
-	logger.Info(context.Background(), "logger", "close and recover")
-	logger.Close()
+	logger.Info(context.Background(), "Stop all", "success", "with input", withInputFlag)
+	// Stop with input first, without input last.
+	if withInputFlag == 0 {
+		logger.Info(context.Background(), "logger", "close and recover")
+		logger.Flush()
+		logger.Close()
+	}
 }
 
 //export Stop

plugin_main/plugin_http.go

@@ -123,10 +123,8 @@ func HandleLoadConfig(w http.ResponseWriter, r *http.Request) {
 func HandleHoldOn(w http.ResponseWriter, r *http.Request) {
 	controlLock.Lock()
 	defer controlLock.Unlock()
-	StopAll(1)
-	StopAll(0)
-	// flush async logs when hold on with exit flag.
-	logger.Flush()
+	StopAllPipelines(1)
+	StopAllPipelines(0)
 	w.WriteHeader(http.StatusOK)
 }
 

plugin_main/plugin_main.go

@@ -176,10 +176,8 @@ func main() {
 	if !*flags.FileIOFlag {
 		<-signals.SetupSignalHandler()
 	}
-	logger.Info(context.Background(), "########################## exit process begin ##########################")
-	StopAll(1)
-	StopAll(0)
-	logger.Info(context.Background(), "########################## exit process done ##########################")
+	StopAllPipelines(1)
+	StopAllPipelines(0)
 }
 
 func generatePluginDoc() {

plugin_main/plugin_main_test.go

@@ -162,8 +162,8 @@ func TestHangConfigWhenStop(t *testing.T) {
 
 	// Quit.
 	time.Sleep(time.Second)
-	StopAll(1)
-	StopAll(0)
+	StopAllPipelines(1)
+	StopAllPipelines(0)
 	pluginmanager.LogtailConfigLock.RLock()
 	require.Empty(t, pluginmanager.LogtailConfig)
 	pluginmanager.LogtailConfigLock.RUnlock()
@@ -225,8 +225,8 @@ func TestSlowConfigWhenStop(t *testing.T) {
 
 	// Quit.
 	time.Sleep(time.Second)
-	StopAll(1)
-	StopAll(0)
+	StopAllPipelines(1)
+	StopAllPipelines(0)
 	time.Sleep(time.Second * 6)
 	require.Empty(t, pluginmanager.DisabledLogtailConfig)
 }

pluginmanager/config_update_test.go

@@ -49,8 +49,8 @@ func (s *configUpdateTestSuite) BeforeTest(suiteName, testName string) {
 
 func (s *configUpdateTestSuite) AfterTest(suiteName, testName string) {
 	logger.Infof(context.Background(), "========== %s %s test end ========================", suiteName, testName)
-	s.NoError(StopAll(false))
-	s.NoError(StopAll(true))
+	s.NoError(StopAllPipelines(false))
+	s.NoError(StopAllPipelines(true))
 	LogtailConfigLock.Lock()
 	LogtailConfig = make(map[string]*LogstoreConfig)
 	LogtailConfigLock.Unlock()
@@ -108,8 +108,8 @@ func (s *configUpdateTestSuite) TestConfigUpdateMany() {
 
 	// load normal config
 	for i := 0; i < 3; i++ {
-		s.NoError(StopAll(true))
-		s.NoError(StopAll(false))
+		s.NoError(StopAllPipelines(true))
+		s.NoError(StopAllPipelines(false))
 		s.NoError(LoadAndStartMockConfig(noblockUpdateConfigName, noblockUpdateConfigName, noblockUpdateConfigName, GetTestConfig(noblockUpdateConfigName)))
 		LogtailConfigLock.RLock()
 		s.NotNil(LogtailConfig[noblockUpdateConfigName])
@@ -163,8 +163,8 @@ func (s *configUpdateTestSuite) TestStopAllExit() {
 	s.True(ok)
 	checkFlusher.Block = false
 	time.Sleep(time.Second * time.Duration(5))
-	s.NoError(StopAll(true))
-	s.NoError(StopAll(false))
+	s.NoError(StopAllPipelines(true))
+	s.NoError(StopAllPipelines(false))
 	s.Equal(20000, checkFlusher.GetLogCount())
 }
 
@@ -177,8 +177,8 @@ func (s *configUpdateTestSuite) TestStopAllExitTimeout() {
 	checkFlusher, ok := GetConfigFlushers(config.PluginRunner)[0].(*checker.FlusherChecker)
 	s.True(ok)
 	s.Equal(0, checkFlusher.GetLogCount())
-	s.NoError(StopAll(true))
-	s.NoError(StopAll(false))
+	s.NoError(StopAllPipelines(true))
+	s.NoError(StopAllPipelines(false))
 	time.Sleep(time.Second)
 	s.Equal(0, checkFlusher.GetLogCount())
 	checkFlusher.Block = false

pluginmanager/plugin_manager.go

@@ -169,10 +169,10 @@ func timeoutStop(config *LogstoreConfig, removedFlag bool) bool {
 	}
 }
 
-// StopAll stops all config instance so that it is ready
+// StopAllPipelines stops all pipelines so that it is ready
 // to quit.
 // For user-defined config, timeoutStop is used to avoid hanging.
-func StopAll(withInput bool) error {
+func StopAllPipelines(withInput bool) error {
 	defer panicRecover("Run plugin")
 
 	LogtailConfigLock.Lock()

plugins/test/common.go

@@ -70,10 +70,10 @@ func PluginStart() error {
 }
 
 func PluginStop() error {
-	if err := pluginmanager.StopAll(true); err != nil {
+	if err := pluginmanager.StopAllPipelines(true); err != nil {
 		return err
 	}
-	if err := pluginmanager.StopAll(false); err != nil {
+	if err := pluginmanager.StopAllPipelines(false); err != nil {
 		return err
 	}
 	return nil