Optimization of the self-monitoring metrics architecture for Go (Golang) plugin modules and unified output of metrics with C++ modules.

core/go_pipeline/LogtailPlugin.cpp

@@ -440,6 +440,12 @@ bool LogtailPlugin::LoadPluginBase() {
             return mPluginValid;
         }
 
+        mGetPipelineMetricsFun = (GetPipelineMetricsFun)loader.LoadMethod("GetPipelineMetrics", error);
+        if (!error.empty()) {
+            LOG_ERROR(sLogger, ("load GetPipelineMetrics error, Message", error));
+            return mPluginValid;
+        }
+
         mPluginBasePtr = loader.Release();
     }
 
@@ -517,6 +523,48 @@ void LogtailPlugin::ProcessLogGroup(const std::string& configName,
     }
 }
 
+void LogtailPlugin::GetPipelineMetrics(std::vector<std::map<std::string, std::string>>& metircsList) {
+    if (mGetPipelineMetricsFun != nullptr) {
+        auto metrics = mGetPipelineMetricsFun();
+        if (metrics != nullptr) {
+            for (int i = 0; i < metrics->count; ++i) {
+                std::map<std::string, std::string> item;
+                InnerPluginMetric* innerpm = metrics->metrics[i];
+                if (innerpm != nullptr) {
+                    for (int j = 0; j < innerpm->count; ++j) {
+                        InnerKeyValue* innerkv = innerpm->keyValues[j];
+                        if (innerkv != nullptr) {
+                            item.insert(std::make_pair(std::string(innerkv->key), std::string(innerkv->value)));                       
+                        }
+                    }
+                }
+                metircsList.emplace_back(item);
+            }
+            if (metrics->count > 0) {
+                for (int i = 0; i < metrics->count; ++i) {
+                    InnerPluginMetric* innerpm = metrics->metrics[i];
+                    if (innerpm != nullptr) {
+                        if (innerpm->count > 0) {
+                            for (int j = 0; j < innerpm->count; ++j) {
+                                InnerKeyValue* innerkv = innerpm->keyValues[j];
+                                if (innerkv != nullptr) {
+                                    free(innerkv->key);
+                                    free(innerkv->value);
+                                    free(innerkv);
+                                }
+                            }
+                            free(innerpm->keyValues);
+                        }
+                        free(innerpm);
+                    }
+                }
+                free(metrics->metrics);
+            }
+            free(metrics);
+        }
+    }
+}
+
 K8sContainerMeta LogtailPlugin::GetContainerMeta(const string& containerID) {
     if (mPluginValid && mGetContainerMetaFun != nullptr) {
         GoString id;

core/go_pipeline/LogtailPlugin.h

@@ -72,6 +72,23 @@ struct innerContainerMeta {
     char** envsKey;
     char** envsVal;
 };
+
+typedef struct {
+    char* key;
+    char* value;
+} InnerKeyValue;
+
+typedef struct {
+    InnerKeyValue** keyValues;
+    int count;
+} InnerPluginMetric;
+
+typedef struct {
+    InnerPluginMetric** metrics;
+    int count;
+} InnerPluginMetrics;
+
+
 struct K8sContainerMeta {
     std::string PodName;
     std::string K8sNamespace;
@@ -122,6 +139,7 @@ typedef GoInt (*InitPluginBaseV2Fun)(GoString cfg);
 typedef GoInt (*ProcessLogsFun)(GoString c, GoSlice l, GoString p, GoString t, GoSlice tags);
 typedef GoInt (*ProcessLogGroupFun)(GoString c, GoSlice l, GoString p);
 typedef struct innerContainerMeta* (*GetContainerMetaFun)(GoString containerID);
+typedef InnerPluginMetrics* (*GetPipelineMetricsFun)();
 
 // Methods export by adapter.
 typedef int (*IsValidToSendFun)(long long logstoreKey);
@@ -234,6 +252,8 @@ class LogtailPlugin {
 
     K8sContainerMeta GetContainerMeta(const std::string& containerID);
 
+    void GetPipelineMetrics(std::vector<std::map<std::string, std::string>>& metircsList);
+
 private:
     void* mPluginBasePtr;
     void* mPluginAdapterPtr;
@@ -252,7 +272,7 @@ class LogtailPlugin {
     ProcessLogsFun mProcessLogsFun;
     ProcessLogGroupFun mProcessLogGroupFun;
     GetContainerMetaFun mGetContainerMetaFun;
-
+    GetPipelineMetricsFun mGetPipelineMetricsFun;
     // Configuration for plugin system in JSON format.
     Json::Value mPluginCfg;
 

core/monitor/LogtaiMetric.cpp

@@ -136,13 +136,15 @@ void WriteMetrics::PreparePluginCommonLabels(const std::string& projectName,
                                              const std::string& configName,
                                              const std::string& pluginName,
                                              const std::string& pluginID,
+                                             const std::string& childPluginID, 
                                              MetricLabels& labels) {
     labels.emplace_back(std::make_pair("project", projectName));
     labels.emplace_back(std::make_pair("logstore", logstoreName));
     labels.emplace_back(std::make_pair("region", region));
     labels.emplace_back(std::make_pair("config_name", configName));
     labels.emplace_back(std::make_pair("plugin_name", pluginName));
     labels.emplace_back(std::make_pair("plugin_id", pluginID));
+    labels.emplace_back(std::make_pair("child_plugin_id", childPluginID));
 }
 
 void WriteMetrics::PrepareMetricsRecordRef(MetricsRecordRef& ref, MetricLabels&& labels) {
@@ -255,7 +257,6 @@ void ReadMetrics::ReadAsLogGroup(std::map<std::string, sls_logs::LogGroup*>& log
     MetricsRecord* tmp = mHead;
     while (tmp) {
         Log* logPtr = nullptr;
-        // for (auto &item: tmp->GetLabels()) {
         for (auto item = tmp->GetLabels()->begin(); item != tmp->GetLabels()->end(); ++item) {
             std::pair<std::string, std::string> pair = *item;
             if (METRIC_FIELD_REGION == pair.first) {

core/monitor/LogtailMetric.h

@@ -101,6 +101,7 @@ class WriteMetrics {
                                    const std::string& configName,
                                    const std::string& pluginName,
                                    const std::string& pluginID,
+                                   const std::string& childPluginID, 
                                    MetricLabels& labels);
     void PrepareMetricsRecordRef(MetricsRecordRef& ref, MetricLabels&& labels);
     MetricsRecord* DoSnapshot();

core/monitor/MetricExportor.cpp

@@ -5,6 +5,8 @@
 #include "config_manager/ConfigManager.h"
 #include "LogFileProfiler.h"
 #include "MetricConstants.h"
+#include "go_pipeline/LogtailPlugin.h"
+#include "app_config/AppConfig.h"
 
 using namespace sls_logs;
 using namespace std;
@@ -13,16 +15,53 @@ namespace logtail {
 
 MetricExportor::MetricExportor() : mSendInterval(60), mLastSendTime(time(NULL) - (rand() % (mSendInterval / 10)) * 10) {}
 
-void MetricExportor::PushMetrics(bool forceSend) {
-    int32_t curTime = time(NULL);
-    if (!forceSend && (curTime - mLastSendTime < mSendInterval)) {
-        return;
+void MetricExportor::PushGoPluginMetrics() {
+    std::vector<std::map<std::string, std::string>> goPluginMetircsList;
+    LogtailPlugin::GetInstance()->GetPipelineMetrics(goPluginMetircsList);
+    std::map<std::string, sls_logs::LogGroup*> goLogGroupMap;
+
+    for (auto& item :  goPluginMetircsList) {
+        std::string configName = "";
+        std::string region = METRIC_REGION_DEFAULT;
+        {
+            // get the config_name label
+            for (const auto& pair : item) {
+                if (pair.first == "label.config_name") {
+                    configName = pair.second;
+                    break;
+                }
+            }
+            if (!configName.empty()) {
+                // get region info by config_name
+                Config* config = ConfigManager::GetInstance()->FindConfigByName(configName);
+                if (config) {
+                    region = config->mRegion;
+                }
+            } 
+        }
+        Log* logPtr = nullptr;
+        auto LogGroupIter = goLogGroupMap.find(region);
+        if (LogGroupIter != goLogGroupMap.end()) {
+            sls_logs::LogGroup* logGroup = LogGroupIter->second;
+            logPtr = logGroup->add_logs();
+        } else {
+            sls_logs::LogGroup* logGroup = new sls_logs::LogGroup();
+            logPtr = logGroup->add_logs();
+            goLogGroupMap.insert(std::pair<std::string, sls_logs::LogGroup*>(region, logGroup));
+        }
+        auto now = GetCurrentLogtailTime();
+        SetLogTime(logPtr,
+                   AppConfig::GetInstance()->EnableLogTimeAutoAdjust() ? now.tv_sec + GetTimeDelta() : now.tv_sec);
+        for (const auto& pair : item) {
+            Log_Content* contentPtr = logPtr->add_contents();
+            contentPtr->set_key(pair.first);
+            contentPtr->set_value(pair.second);
+        }   
     }
-    ReadMetrics::GetInstance()->UpdateMetrics();
-
-    std::map<std::string, sls_logs::LogGroup*> logGroupMap;
-    ReadMetrics::GetInstance()->ReadAsLogGroup(logGroupMap);
+    SendMetrics(goLogGroupMap);
+}
 
+void MetricExportor::SendMetrics(std::map<std::string, sls_logs::LogGroup*>& logGroupMap) {
     std::map<std::string, sls_logs::LogGroup*>::iterator iter;
     for (iter = logGroupMap.begin(); iter != logGroupMap.end(); iter ++) {
         sls_logs::LogGroup* logGroup = iter->second;
@@ -37,4 +76,21 @@ void MetricExportor::PushMetrics(bool forceSend) {
         delete logGroup;
     }
 }
+
+void MetricExportor::PushMetrics(bool forceSend) {
+    int32_t curTime = time(NULL);
+    if (!forceSend && (curTime - mLastSendTime < mSendInterval)) {
+        return;
+    }
+
+    if (LogtailPlugin::GetInstance()->IsPluginOpened()) {
+        PushGoPluginMetrics();
+    }
+
+    ReadMetrics::GetInstance()->UpdateMetrics();    
+    std::map<std::string, sls_logs::LogGroup*> logGroupMap;
+    ReadMetrics::GetInstance()->ReadAsLogGroup(logGroupMap);
+
+    SendMetrics(logGroupMap);
+}
 }

core/monitor/MetricExportor.h

@@ -12,6 +12,8 @@ class MetricExportor {
         return ptr;
     }
     void PushMetrics(bool forceSend);
+    void PushGoPluginMetrics();
+    void SendMetrics(std::map<std::string, sls_logs::LogGroup*>& logGroupMap);
 
 private:
     MetricExportor();

core/pipeline/Pipeline.cpp

@@ -28,9 +28,21 @@
 #include "processor/ProcessorDesensitizeNative.h"
 #include "processor/ProcessorTagNative.h"
 #include "processor/ProcessorFilterNative.h"
+#include "plugin/instance/PluginInstance.h"
+
 
 namespace logtail {
 
+void genPluginAndNodeID(int& pluginIndex, bool lastOne, PluginInstance::PluginMeta& pluginMeta) {
+    pluginIndex ++;
+    int childPluginID = pluginIndex;
+    if (!lastOne) {
+        childPluginID = pluginIndex + 1;
+    }
+    pluginMeta.pluginID = std::to_string(pluginIndex);
+    pluginMeta.childPluginID = std::to_string(childPluginID);
+}
+
 bool Pipeline::Init(const PipelineConfig& config) {
     mName = config.mConfigName;
     mConfig = config;
@@ -41,15 +53,18 @@ bool Pipeline::Init(const PipelineConfig& config) {
     mContext.SetRegion(config.mRegion);
 
     int pluginIndex = 0;
+
+    PluginInstance::PluginMeta pluginMeta;
     // Input plugin
     pluginIndex++;
 
     if (config.mLogType == STREAM_LOG || config.mLogType == PLUGIN_LOG) {
         return true;
     }
 
+    genPluginAndNodeID(pluginIndex, false, pluginMeta);
     std::unique_ptr<ProcessorInstance> pluginGroupInfo = PluginRegistry::GetInstance()->CreateProcessor(
-        ProcessorTagNative::sName, std::to_string(pluginIndex++));
+        ProcessorTagNative::sName, pluginMeta);
     if (!InitAndAddProcessor(std::move(pluginGroupInfo), config)) {
         return false;
     }
@@ -59,35 +74,37 @@ bool Pipeline::Init(const PipelineConfig& config) {
     }
 
     std::unique_ptr<ProcessorInstance> pluginDecoder;
+    genPluginAndNodeID(pluginIndex, false, pluginMeta);
     if (config.mLogType == JSON_LOG || !config.IsMultiline()) {
         pluginDecoder = PluginRegistry::GetInstance()->CreateProcessor(ProcessorSplitLogStringNative::sName,
-                                                                       std::to_string(pluginIndex++));
+                                                                       pluginMeta);
     } else {
         pluginDecoder = PluginRegistry::GetInstance()->CreateProcessor(ProcessorSplitRegexNative::sName,
-                                                                       std::to_string(pluginIndex++));
+                                                                       pluginMeta);
     }
     if (!InitAndAddProcessor(std::move(pluginDecoder), config)) {
         return false;
     }
 
     // APSARA_LOG, REGEX_LOG, STREAM_LOG, JSON_LOG, DELIMITER_LOG, PLUGIN_LOG
     std::unique_ptr<ProcessorInstance> pluginParser;
+    genPluginAndNodeID(pluginIndex, false, pluginMeta);
     switch (config.mLogType) {
         case APSARA_LOG:
             pluginParser = PluginRegistry::GetInstance()->CreateProcessor(ProcessorParseApsaraNative::sName,
-                                                                          std::to_string(pluginIndex++));
+                                                                          pluginMeta);
             break;
         case REGEX_LOG:
             pluginParser = PluginRegistry::GetInstance()->CreateProcessor(ProcessorParseRegexNative::sName,
-                                                                          std::to_string(pluginIndex++));
+                                                                          pluginMeta);
             break;
         case JSON_LOG:
             pluginParser = PluginRegistry::GetInstance()->CreateProcessor(ProcessorParseJsonNative::sName,
-                                                                          std::to_string(pluginIndex++));
+                                                                          pluginMeta);
             break;
         case DELIMITER_LOG:
             pluginParser = PluginRegistry::GetInstance()->CreateProcessor(ProcessorParseDelimiterNative::sName,
-                                                                          std::to_string(pluginIndex++));
+                                                                          pluginMeta);
             break;
         default:
             return false;
@@ -96,21 +113,28 @@ bool Pipeline::Init(const PipelineConfig& config) {
         return false;
     }
 
+    genPluginAndNodeID(pluginIndex, false, pluginMeta);
     std::unique_ptr<ProcessorInstance> pluginTime = PluginRegistry::GetInstance()->CreateProcessor(
-        ProcessorParseTimestampNative::sName, std::to_string(pluginIndex++));
+        ProcessorParseTimestampNative::sName, pluginMeta);
     if (!InitAndAddProcessor(std::move(pluginTime), config)) {
         return false;
     }
 
+    if (!config.mSensitiveWordCastOptions.empty()) {
+        genPluginAndNodeID(pluginIndex, false, pluginMeta);
+    } else {
+        genPluginAndNodeID(pluginIndex, true, pluginMeta);
+    }
     std::unique_ptr<ProcessorInstance> pluginFilter = PluginRegistry::GetInstance()->CreateProcessor(
-        ProcessorFilterNative::sName, std::to_string(pluginIndex++));
+        ProcessorFilterNative::sName, pluginMeta);
     if (!InitAndAddProcessor(std::move(pluginFilter), config)) {
         return false;
     }
 
     if (!config.mSensitiveWordCastOptions.empty()) {
+        genPluginAndNodeID(pluginIndex, true, pluginMeta);
         std::unique_ptr<ProcessorInstance> pluginDesensitize = PluginRegistry::GetInstance()->CreateProcessor(
-            ProcessorDesensitizeNative::sName, std::to_string(pluginIndex++));
+            ProcessorDesensitizeNative::sName, pluginMeta);
         if (!InitAndAddProcessor(std::move(pluginDesensitize), config)) {
             return false;
         }
@@ -131,9 +155,9 @@ bool Pipeline::InitAndAddProcessor(std::unique_ptr<ProcessorInstance>&& processo
                   ("CreateProcessor", ProcessorSplitRegexNative::sName)("Error", "Cannot find plugin"));
         return false;
     }
-    ComponentConfig componentConfig(processor->Id(), config);
+    ComponentConfig componentConfig(processor->Meta().pluginID, config);
     if (!processor->Init(componentConfig, mContext)) {
-        LOG_ERROR(GetContext().GetLogger(), ("InitProcessor", processor->Id())("Error", "Init failed"));
+        LOG_ERROR(GetContext().GetLogger(), ("InitProcessor", processor->Meta().pluginID)("Error", "Init failed"));
         return false;
     }
     mProcessorLine.emplace_back(std::move(processor));

core/plugin/PluginRegistry.cpp

@@ -69,17 +69,17 @@ void PluginRegistry::UnloadPlugins() {
 }
 
 std::unique_ptr<ProcessorInstance> PluginRegistry::CreateProcessor(const std::string& name,
-                                                                   const std::string& pluginId) {
+                                                                   const PluginInstance::PluginMeta& pluginMeta) {
     return std::unique_ptr<ProcessorInstance>(
-        static_cast<ProcessorInstance*>(Create(PROCESSOR_PLUGIN, name, pluginId).release()));
+        static_cast<ProcessorInstance*>(Create(PROCESSOR_PLUGIN, name, pluginMeta).release()));
 }
 
 std::unique_ptr<PluginInstance>
-PluginRegistry::Create(PluginCat cat, const std::string& name, const std::string& pluginId) {
+PluginRegistry::Create(PluginCat cat, const std::string& name, const PluginInstance::PluginMeta& pluginMeta) {
     std::unique_ptr<PluginInstance> ins;
     auto creatorEntry = mPluginDict.find(PluginKey(cat, name));
     if (creatorEntry != mPluginDict.end()) {
-        ins = creatorEntry->second->Create(pluginId);
+        ins = creatorEntry->second->Create(pluginMeta);
     }
     return ins;
 }