Merge branch 'main' into feat/prom-tls-curl

alibaba · Nov 25, 2024 · c57ff19 · c57ff19
2 parents 8c27ef3 + b11fd8f
commit c57ff19
Show file tree

Hide file tree

Showing 10 changed files with 62 additions and 22 deletions.
diff --git a/core/app_config/AppConfig.cpp b/core/app_config/AppConfig.cpp
@@ -25,6 +25,7 @@
 #include "common/FileSystemUtil.h"
 #include "common/JsonUtil.h"
 #include "common/LogtailCommonFlags.h"
+#include "common/version.h"
 #include "config/InstanceConfigManager.h"
 #include "config/watcher/InstanceConfigWatcher.h"
 #include "file_server/ConfigManager.h"
@@ -39,6 +40,10 @@
 
 using namespace std;
 
+#define ILOGTAIL_PREFIX "ilogtail_"
+#define ILOGTAIL_PIDFILE_SUFFIX ".pid"
+#define LOONGCOLLECTOR_PREFIX "loongcollector_"
+
 DEFINE_FLAG_BOOL(logtail_mode, "logtail mode", false);
 DEFINE_FLAG_INT32(max_buffer_num, "max size", 40);
 DEFINE_FLAG_INT32(pub_max_buffer_num, "max size", 8);
@@ -520,7 +525,7 @@ std::string GetAgentName() {
         return "ilogtail";
     } else {
         return "loongcollector";
-    } 
+    }
 }
 
 std::string GetMonitorInfoFileName() {
@@ -531,6 +536,30 @@ std::string GetMonitorInfoFileName() {
     }
 }
 
+std::string GetSymLinkName() {
+    if (BOOL_FLAG(logtail_mode)) {
+        return GetProcessExecutionDir() + "ilogtail";
+    } else {
+        return GetProcessExecutionDir() + "loongcollector";
+    }
+}
+
+std::string GetPidFileName() {
+    if (BOOL_FLAG(logtail_mode)) {
+        return GetProcessExecutionDir() + ILOGTAIL_PREFIX + ILOGTAIL_VERSION + ILOGTAIL_PIDFILE_SUFFIX;
+    } else {
+        return GetAgentRunDir() + "loongcollector.pid";
+    }
+}
+
+std::string GetAgentPrefix() {
+    if (BOOL_FLAG(logtail_mode)) {
+        return ILOGTAIL_PREFIX;
+    } else {
+        return LOONGCOLLECTOR_PREFIX;
+    }
+}
+
 AppConfig::AppConfig() {
     LOG_INFO(sLogger, ("AppConfig AppConfig", "success"));
     SetIlogtailConfigJson("");
@@ -1509,8 +1538,8 @@ void AppConfig::ReadFlagsFromMap(const std::unordered_map<std::string, std::stri
  *    - 记录无法转换的值
  */
 void AppConfig::RecurseParseJsonToFlags(const Json::Value& confJson, std::string prefix) {
-    const static unordered_set<string> sIgnoreKeySet = {"data_server_list", "legacy_data_server_list"};
-    const static unordered_set<string> sForceKeySet = {"config_server_address_list", "config_server_list"};
+    const static unordered_set<string> sIgnoreKeySet = {"data_server_list", "data_servers"};
+    const static unordered_set<string> sForceKeySet = {"config_server_address_list", "config_servers"};
     for (auto name : confJson.getMemberNames()) {
         auto jsonvalue = confJson[name];
         string fullName;

diff --git a/core/app_config/AppConfig.h b/core/app_config/AppConfig.h
@@ -61,6 +61,9 @@ std::string GetVersionTag();
 std::string GetGoPluginCheckpoint();
 std::string GetAgentName();
 std::string GetMonitorInfoFileName();
+std::string GetSymLinkName();
+std::string GetPidFileName();
+std::string GetAgentPrefix();
 
 template <class T>
 class DoubleBuffer {

diff --git a/core/common/version.h b/core/common/version.h
@@ -22,7 +22,4 @@ extern const char* const ILOGTAIL_UPDATE_SUFFIX;
 extern const char* const ILOGTAIL_GIT_HASH;
 extern const char* const ILOGTAIL_BUILD_DATE;
 
-#define ILOGTAIL_PREFIX "ilogtail_"
-#define ILOGTAIL_PIDFILE_SUFFIX ".pid"
-
 #endif
diff --git a/core/pipeline/Pipeline.cpp b/core/pipeline/Pipeline.cpp
@@ -167,7 +167,7 @@ bool Pipeline::Init(PipelineConfig&& config) {
             = PluginRegistry::GetInstance()->CreateFlusher(pluginType, GenNextPluginMeta(false));
         if (flusher) {
             Json::Value optionalGoPipeline;
-            if (!flusher->Init(detail, mContext, optionalGoPipeline)) {
+            if (!flusher->Init(detail, mContext, i, optionalGoPipeline)) {
                 return false;
             }
             mFlushers.emplace_back(std::move(flusher));

diff --git a/core/pipeline/batch/Batcher.h b/core/pipeline/batch/Batcher.h
@@ -176,7 +176,7 @@ class Batcher {
                         }
                         if (mGroupQueue->IsEmpty()) {
                             TimeoutFlushManager::GetInstance()->UpdateRecord(mFlusher->GetContext().GetConfigName(),
-                                                                             0,
+                                                                             mFlusher->GetFlusherIndex(),
                                                                              0,
                                                                              mGroupFlushStrategy->GetTimeoutSecs(),
                                                                              mFlusher);
@@ -193,8 +193,11 @@ class Batcher {
                                g.GetSourceBuffer(),
                                g.GetExactlyOnceCheckpoint(),
                                g.GetMetadata(EventGroupMetaKey::SOURCE_ID));
-                    TimeoutFlushManager::GetInstance()->UpdateRecord(
-                        mFlusher->GetContext().GetConfigName(), 0, key, mEventFlushStrategy.GetTimeoutSecs(), mFlusher);
+                    TimeoutFlushManager::GetInstance()->UpdateRecord(mFlusher->GetContext().GetConfigName(),
+                                                                     mFlusher->GetFlusherIndex(),
+                                                                     key,
+                                                                     mEventFlushStrategy.GetTimeoutSecs(),
+                                                                     mFlusher);
                     mBufferedGroupsTotal->Add(1);
                     mBufferedDataSizeByte->Add(item.DataSize());
                 } else if (i == 0) {
@@ -243,8 +246,11 @@ class Batcher {
             mGroupQueue->Flush(res);
         }
         if (mGroupQueue->IsEmpty()) {
-            TimeoutFlushManager::GetInstance()->UpdateRecord(
-                mFlusher->GetContext().GetConfigName(), 0, 0, mGroupFlushStrategy->GetTimeoutSecs(), mFlusher);
+            TimeoutFlushManager::GetInstance()->UpdateRecord(mFlusher->GetContext().GetConfigName(),
+                                                             mFlusher->GetFlusherIndex(),
+                                                             0,
+                                                             mGroupFlushStrategy->GetTimeoutSecs(),
+                                                             mFlusher);
         }
         iter->second.Flush(mGroupQueue.value());
         mEventQueueMap.erase(iter);

diff --git a/core/pipeline/plugin/instance/FlusherInstance.cpp b/core/pipeline/plugin/instance/FlusherInstance.cpp
@@ -20,9 +20,10 @@ using namespace std;
 
 namespace logtail {
 
-bool FlusherInstance::Init(const Json::Value& config, PipelineContext& context, Json::Value& optionalGoPipeline) {
+bool FlusherInstance::Init(const Json::Value& config, PipelineContext& context, size_t flusherIdx, Json::Value& optionalGoPipeline) {
     mPlugin->SetContext(context);
     mPlugin->SetPluginID(PluginID());
+    mPlugin->SetFlusherIndex(flusherIdx);
     mPlugin->SetMetricsRecordRef(Name(), PluginID());
     if (!mPlugin->Init(config, optionalGoPipeline)) {
         return false;

diff --git a/core/pipeline/plugin/instance/FlusherInstance.h b/core/pipeline/plugin/instance/FlusherInstance.h
@@ -31,12 +31,13 @@ namespace logtail {
 
 class FlusherInstance : public PluginInstance {
 public:
-    FlusherInstance(Flusher* plugin, const PluginInstance::PluginMeta& pluginMeta) : PluginInstance(pluginMeta), mPlugin(plugin) {}
+    FlusherInstance(Flusher* plugin, const PluginInstance::PluginMeta& pluginMeta)
+        : PluginInstance(pluginMeta), mPlugin(plugin) {}
 
     const std::string& Name() const override { return mPlugin->Name(); };
     const Flusher* GetPlugin() const { return mPlugin.get(); }
 
-    bool Init(const Json::Value& config, PipelineContext& context, Json::Value& optionalGoPipeline);
+    bool Init(const Json::Value& config, PipelineContext& context, size_t flusherIdx, Json::Value& optionalGoPipeline);
     bool Start() { return mPlugin->Start(); }
     bool Stop(bool isPipelineRemoving) { return mPlugin->Stop(isPipelineRemoving); }
     bool Send(PipelineEventGroup&& g);

diff --git a/core/pipeline/plugin/interface/Flusher.h b/core/pipeline/plugin/interface/Flusher.h
@@ -44,6 +44,8 @@ class Flusher : public Plugin {
 
     QueueKey GetQueueKey() const { return mQueueKey; }
     void SetPluginID(const std::string& pluginID) { mPluginID = pluginID; }
+    size_t GetFlusherIndex() { return mIndex; }
+    void SetFlusherIndex(size_t idx) { mIndex = idx; }
     const std::string& GetPluginID() const { return mPluginID; }
 
 protected:
@@ -54,6 +56,7 @@ class Flusher : public Plugin {
 
     QueueKey mQueueKey;
     std::string mPluginID;
+    size_t mIndex = 0;
 
 #ifdef APSARA_UNIT_TEST_MAIN
     friend class FlusherInstanceUnittest;

diff --git a/core/unittest/pipeline/PipelineUnittest.cpp b/core/unittest/pipeline/PipelineUnittest.cpp
@@ -2736,13 +2736,13 @@ void PipelineUnittest::TestSend() const {
         {
             auto flusher
                 = PluginRegistry::GetInstance()->CreateFlusher(FlusherMock::sName, pipeline.GenNextPluginMeta(false));
-            flusher->Init(Json::Value(), ctx, tmp);
+            flusher->Init(Json::Value(), ctx, 0, tmp);
             pipeline.mFlushers.emplace_back(std::move(flusher));
         }
         {
             auto flusher
                 = PluginRegistry::GetInstance()->CreateFlusher(FlusherMock::sName, pipeline.GenNextPluginMeta(false));
-            flusher->Init(Json::Value(), ctx, tmp);
+            flusher->Init(Json::Value(), ctx, 0, tmp);
             pipeline.mFlushers.emplace_back(std::move(flusher));
         }
         vector<pair<size_t, const Json::Value*>> configs;
@@ -2788,13 +2788,13 @@ void PipelineUnittest::TestSend() const {
         {
             auto flusher
                 = PluginRegistry::GetInstance()->CreateFlusher(FlusherMock::sName, pipeline.GenNextPluginMeta(false));
-            flusher->Init(Json::Value(), ctx, tmp);
+            flusher->Init(Json::Value(), ctx, 0, tmp);
             pipeline.mFlushers.emplace_back(std::move(flusher));
         }
         {
             auto flusher
                 = PluginRegistry::GetInstance()->CreateFlusher(FlusherMock::sName, pipeline.GenNextPluginMeta(false));
-            flusher->Init(Json::Value(), ctx, tmp);
+            flusher->Init(Json::Value(), ctx, 0, tmp);
             pipeline.mFlushers.emplace_back(std::move(flusher));
         }
 
@@ -2855,13 +2855,13 @@ void PipelineUnittest::TestFlushBatch() const {
     {
         auto flusher
             = PluginRegistry::GetInstance()->CreateFlusher(FlusherMock::sName, pipeline.GenNextPluginMeta(false));
-        flusher->Init(Json::Value(), ctx, tmp);
+        flusher->Init(Json::Value(), ctx, 0, tmp);
         pipeline.mFlushers.emplace_back(std::move(flusher));
     }
     {
         auto flusher
             = PluginRegistry::GetInstance()->CreateFlusher(FlusherMock::sName, pipeline.GenNextPluginMeta(false));
-        flusher->Init(Json::Value(), ctx, tmp);
+        flusher->Init(Json::Value(), ctx, 0, tmp);
         pipeline.mFlushers.emplace_back(std::move(flusher));
     }
     {

diff --git a/core/unittest/plugin/FlusherInstanceUnittest.cpp b/core/unittest/plugin/FlusherInstanceUnittest.cpp
@@ -46,7 +46,7 @@ void FlusherInstanceUnittest::TestInit() const {
         = make_unique<FlusherInstance>(new FlusherMock(), PluginInstance::PluginMeta("0"));
     Json::Value config, opt;
     PipelineContext context;
-    APSARA_TEST_TRUE(flusher->Init(config, context, opt));
+    APSARA_TEST_TRUE(flusher->Init(config, context, 0, opt));
     APSARA_TEST_EQUAL(&context, &flusher->GetPlugin()->GetContext());
 }