fix

alibaba · Dec 27, 2024 · 88d8dae · 88d8dae
1 parent 3b542a0
commit 88d8dae
Show file tree

Hide file tree

Showing 6 changed files with 158 additions and 29 deletions.
diff --git a/core/config/PipelineConfig.cpp b/core/config/PipelineConfig.cpp
@@ -246,11 +246,13 @@ bool PipelineConfig::Parse() {
             }
         }
         mInputs.push_back(&plugin);
+#ifndef APSARA_UNIT_TEST_MAIN
         // TODO: remove these special restrictions
         if (pluginType == "input_file" || pluginType == "input_container_stdio") {
             hasFileInput = true;
         }
-#ifdef APSARA_UNIT_TEST_MAIN
+#else
+        // TODO: remove these special restrictions after all C++ inputs support Go processors
         if (pluginType.find("input_file") != string::npos || pluginType.find("input_container_stdio") != string::npos) {
             hasFileInput = true;
         }

diff --git a/core/monitor/profile_sender/ProfileSender.cpp b/core/monitor/profile_sender/ProfileSender.cpp
@@ -119,6 +119,7 @@ FlusherSLS* ProfileSender::GetFlusher(const string& region) {
 }
 
 bool ProfileSender::IsProfileData(const string& region, const string& project, const string& logstore) {
+// TODO: temporarily used, profile should work in unit test
 #ifndef APSARA_UNIT_TEST_MAIN
     if ((logstore == "shennong_log_profile" || logstore == "logtail_alarm" || logstore == "logtail_status_profile"
          || logstore == "logtail_suicide_profile")

diff --git a/core/pipeline/Pipeline.cpp b/core/pipeline/Pipeline.cpp
@@ -395,9 +395,8 @@ bool Pipeline::Send(vector<PipelineEventGroup>&& groupList) {
         auto res = mRouter.Route(group);
         for (auto& item : res) {
             if (item.first >= mFlushers.size()) {
-                LOG_WARNING(sLogger,
-                            ("pipeline send", "discard data")("config", mName)(
-                                "reason", "invalid flusher index or config update flusher from C++ to Go"));
+                LOG_ERROR(sLogger,
+                          ("unexpected error", "invalid flusher index")("flusher index", item.first)("config", mName));
                 allSucceeded = false;
                 continue;
             }

diff --git a/core/pipeline/Pipeline.h b/core/pipeline/Pipeline.h
@@ -62,7 +62,6 @@ class Pipeline {
     PipelineContext& GetContext() const { return mContext; }
     const Json::Value& GetConfig() const { return *mConfig; }
     const std::optional<std::string>& GetSingletonInput() const { return mSingletonInput; }
-    const std::vector<std::unique_ptr<ProcessorInstance>>& GetProcessors() const { return mProcessorLine; }
     const std::vector<std::unique_ptr<FlusherInstance>>& GetFlushers() const { return mFlushers; }
     bool IsFlushingThroughGoPipeline() const { return !mGoPipelineWithoutInput.isNull(); }
     const std::unordered_map<std::string, std::unordered_map<std::string, uint32_t>>& GetPluginStatistics() const {
@@ -129,6 +128,7 @@ class Pipeline {
     friend class InputProcessSecurityUnittest;
     friend class InputNetworkSecurityUnittest;
     friend class InputNetworkObserverUnittest;
+    friend class PipelineUpdateUnittest;
 #endif
 };
 

diff --git a/core/pipeline/plugin/instance/ProcessorInstance.h b/core/pipeline/plugin/instance/ProcessorInstance.h
@@ -33,7 +33,6 @@ class ProcessorInstance : public PluginInstance {
     ProcessorInstance(Processor* plugin, const PluginMeta& pluginMeta) : PluginInstance(pluginMeta), mPlugin(plugin) {}
 
     const std::string& Name() const override { return mPlugin->Name(); };
-    const Processor* GetPlugin() const { return mPlugin.get(); }
 
     bool Init(const Json::Value& config, PipelineContext& context);
     void Process(std::vector<PipelineEventGroup>& logGroupList);
@@ -60,6 +59,7 @@ class ProcessorInstance : public PluginInstance {
     friend class InputFileUnittest;
     friend class InputPrometheusUnittest;
     friend class PipelineUnittest;
+    friend class PipelineUpdateUnittest;
 #endif
 };