Feat: Reporting metrics upon first startup

core/application/Application.cpp

@@ -276,6 +276,21 @@ void Application::Start() { // GCOVR_EXCL_START
 
     OnetimeConfigInfoManager::GetInstance()->LoadCheckpointFile();
 
+#ifdef __ENTERPRISE__
+    // Wait for all runners to be started, then load built-in pipelines first, followed by sending start metrics.
+    auto configDiff = PipelineConfigWatcher::GetInstance()->CheckConfigDiff(true);
+    if (!configDiff.first.IsEmpty()) {
+        CollectionPipelineManager::GetInstance()->UpdatePipelines(configDiff.first);
+    }
+    if (!configDiff.second.IsEmpty()) {
+        TaskPipelineManager::GetInstance()->UpdatePipelines(configDiff.second);
+    }
+    if (!configDiff.first.IsEmpty() || !configDiff.second.IsEmpty()) {
+        OnetimeConfigInfoManager::GetInstance()->DumpCheckpointFile();
+    }
+    LoongCollectorMonitor::GetInstance()->SendStartMetric();
+#endif
+
     time_t curTime = 0, lastOnetimeConfigTimeoutCheckTime = 0, lastConfigCheckTime = 0, lastUpdateMetricTime = 0,
            lastCheckTagsTime = 0, lastQueueGCTime = 0, lastCheckUnusedCheckpointsTime = 0;
     while (true) {

core/config/watcher/PipelineConfigWatcher.cpp

@@ -37,15 +37,18 @@ PipelineConfigWatcher::PipelineConfigWatcher()
       mTaskPipelineManager(TaskPipelineManager::GetInstance()) {
 }
 
-pair<CollectionConfigDiff, TaskConfigDiff> PipelineConfigWatcher::CheckConfigDiff() {
+pair<CollectionConfigDiff, TaskConfigDiff> PipelineConfigWatcher::CheckConfigDiff(bool builtinOnly) {
     CollectionConfigDiff pDiff;
     TaskConfigDiff tDiff;
     unordered_set<string> configSet;
     SingletonConfigCache singletonCache;
     // builtin pipeline configs
     InsertBuiltInPipelines(pDiff, tDiff, configSet, singletonCache);
-    // file pipeline configs
-    InsertPipelines(pDiff, tDiff, configSet, singletonCache);
+
+    if (!builtinOnly) {
+        // file pipeline configs
+        InsertPipelines(pDiff, tDiff, configSet, singletonCache);
+    }
 
     CheckSingletonInput(pDiff, singletonCache);
     for (const auto& name : mCollectionPipelineManager->GetAllConfigNames()) {
@@ -100,6 +103,13 @@ void PipelineConfigWatcher::InsertBuiltInPipelines(CollectionConfigDiff& pDiff,
     for (const auto& pipeline : builtInPipelines) {
         const string& pipelineName = pipeline.first;
         const string& pipleineDetail = pipeline.second;
+        if (configSet.find(pipelineName) != configSet.end()) {
+            LOG_WARNING(sLogger,
+                        ("more than 1 config with the same name is found", "skip current config")("inner pipeline",
+                                                                                                  pipelineName));
+            continue;
+        }
+        configSet.insert(pipelineName);
 
         string errorMsg;
         auto iter = mInnerConfigMap.find(pipelineName);

core/config/watcher/PipelineConfigWatcher.h

@@ -48,7 +48,7 @@ class PipelineConfigWatcher : public ConfigWatcher {
         return &instance;
     }
 
-    std::pair<CollectionConfigDiff, TaskConfigDiff> CheckConfigDiff();
+    std::pair<CollectionConfigDiff, TaskConfigDiff> CheckConfigDiff(bool builtinOnly = false);
 
 #ifdef APSARA_UNIT_TEST_MAIN
     void SetPipelineManager(const CollectionPipelineManager* pm) { mCollectionPipelineManager = pm; }

core/monitor/Monitor.cpp

@@ -644,6 +644,10 @@ void LoongCollectorMonitor::Stop() {
     LOG_INFO(sLogger, ("LoongCollector monitor", "stopped successfully"));
 }
 
+void LoongCollectorMonitor::SendStartMetric() {
+    SelfMonitorServer::GetInstance()->SendStartMetric();
+}
+
 bool LoongCollectorMonitor::GetAgentMetric(SelfMonitorMetricEvent& event) {
     lock_guard<mutex> lock(mGlobalMetricsMux);
     event = mGlobalMetrics.mAgentMetric;
@@ -675,4 +679,14 @@ void LoongCollectorMonitor::SetRunnerMetric(const std::string& runnerName, const
     mGlobalMetrics.mRunnerMetrics[runnerName] = event;
 }
 
+void LogtailMonitor::UpdateCpuMem() {
+    CpuStat curCpuStat;
+    if (GetCpuStat(curCpuStat)) {
+        GetMemStat();
+        LoongCollectorMonitor::GetInstance()->SetAgentMemory(mMemStat.mRss);
+        CalCpuStat(curCpuStat, mCpuStat);
+        LoongCollectorMonitor::GetInstance()->SetAgentCpu(mCpuStat.mCpuUsage);
+    }
+}
+
 } // namespace logtail

core/monitor/Monitor.h

@@ -97,6 +97,10 @@ class LogtailMonitor {
     // LogInput use it to do flow control.
     float GetRealtimeCpuLevel() { return mRealtimeCpuStat.mCpuUsage / mScaledCpuUsageUpLimit; }
 
+
+    // UpdateCpuMem updates CPU and Memory statistics and sets them to LoongCollectorMonitor
+    void UpdateCpuMem();
+
 private:
     LogtailMonitor();
     ~LogtailMonitor() = default;
@@ -192,6 +196,7 @@ class LoongCollectorMonitor {
 
     void Init();
     void Stop();
+    void SendStartMetric();
 
     bool GetAgentMetric(SelfMonitorMetricEvent& event);
     void SetAgentMetric(const SelfMonitorMetricEvent& event);

core/monitor/SelfMonitorServer.cpp

@@ -16,8 +16,14 @@
 
 #include "monitor/SelfMonitorServer.h"
 
+#include <json/json.h>
+
+#include <fstream>
+
 #include "MetricConstants.h"
 #include "Monitor.h"
+#include "app_config/AppConfig.h"
+#include "common/FileSystemUtil.h"
 #include "runner/ProcessorRunner.h"
 
 using namespace std;
@@ -100,7 +106,7 @@ void SelfMonitorServer::RemoveMetricPipeline() {
     LOG_INFO(sLogger, ("self-monitor metrics pipeline", "removed"));
 }
 
-void SelfMonitorServer::SendMetrics() {
+void SelfMonitorServer::SendMetrics(bool updateAppInfo) {
     ReadMetrics::GetInstance()->UpdateMetrics();
 
     ReadLock lock(mMetricPipelineLock);
@@ -117,6 +123,10 @@ void SelfMonitorServer::SendMetrics() {
     pipelineEventGroup.SetMetadata(EventGroupMetaKey::INTERNAL_DATA_TYPE, INTERNAL_DATA_TYPE_METRIC);
     ReadAsPipelineEventGroup(pipelineEventGroup);
 
+    if (updateAppInfo) {
+        UpdateAppInfoJson(pipelineEventGroup);
+    }
+
     if (pipelineEventGroup.GetEvents().size() > 0) {
         ProcessorRunner::GetInstance()->PushQueue(
             mMetricPipelineCtx->GetProcessQueueKey(), mMetricInputIndex, std::move(pipelineEventGroup));
@@ -213,4 +223,79 @@ void SelfMonitorServer::SendAlarms() {
     }
 }
 
+void SelfMonitorServer::SendStartMetric() {
+    // 等待mMetricPipelineCtx存在，最多等待1秒
+    auto startTime = std::chrono::steady_clock::now();
+    while (true) {
+        ReadLock lock(mMetricPipelineLock);
+        if (mMetricPipelineCtx != nullptr && mSelfMonitorMetricRules != nullptr) {
+            auto endTime = std::chrono::steady_clock::now();
+            auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);
+            LOG_INFO(sLogger,
+                     ("send start metric", "wait metric pipeline ready")("cost milliseconds", duration.count()));
+            break;
+        }
+        lock.unlock();
+
+        // 检查是否超时
+        auto currentTime = std::chrono::steady_clock::now();
+        auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(currentTime - startTime);
+        if (elapsed.count() >= 1000) { // 1秒超时，实测正常启动应该1ms内完成
+            LOG_WARNING(sLogger, ("send start metric", "wait metric pipeline ready timeout, skip"));
+            return;
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    // 启动时需要立刻更新一下cpu和memory指标，因为正常启动时，cpu和memory指标会等30s更新一次
+    LogtailMonitor::GetInstance()->UpdateCpuMem();
+
+    // 调用SendMetrics，并启用UpdateAppInfo
+    // TODO: SendMetrics(true);
+    SendMetrics();
+    LOG_INFO(sLogger, ("send start metric", "sent successfully"));
+}
+
+void SelfMonitorServer::UpdateAppInfoJson(const PipelineEventGroup& pipelineEventGroup) {
+    // 读取现有的app_info.json
+    string appInfoFile = GetAgentAppInfoFile();
+    Json::Value appInfoJson;
+
+    // 尝试读取现有文件
+    ifstream file(appInfoFile);
+    if (file.is_open()) {
+        Json::CharReaderBuilder builder;
+        string errors;
+        if (Json::parseFromStream(builder, file, &appInfoJson, &errors)) {
+            file.close();
+        } else {
+            file.close();
+            LOG_WARNING(sLogger, ("failed to parse existing app_info.json", errors));
+        }
+    }
+
+    const auto& events = pipelineEventGroup.GetEvents();
+    for (const auto& eventPtr : events) {
+        if (!eventPtr.Is<MetricEvent>()) {
+            continue;
+        }
+
+        const auto& metricEvent = eventPtr.Cast<MetricEvent>();
+        const std::string metricName = metricEvent.GetName().to_string();
+
+        // 检查是否是agent指标
+        if (metricName == MetricCategory::METRIC_CATEGORY_AGENT) {
+            // TODO: 从metricEvent中提取agent指标设置到appInfoJson
+        }
+        // 检查是否是runner指标
+        else if (metricName == MetricCategory::METRIC_CATEGORY_RUNNER) {
+            // TODO: 从metricEvent中提取runner指标，例如所有已经启动的runner的名字
+        }
+    }
+    // 写入更新后的app_info.json
+    string appInfo = appInfoJson.toStyledString();
+    OverwriteFile(appInfoFile, appInfo);
+    LOG_INFO(sLogger, ("updated app_info.json with additional metric info", appInfo));
+}
+
 } // namespace logtail

core/monitor/SelfMonitorServer.h

@@ -38,6 +38,11 @@ class SelfMonitorServer {
     void UpdateAlarmPipeline(CollectionPipelineContext* ctx, size_t inputIndex);
     void RemoveAlarmPipeline();
 
+    // send start metric (agent/runner)
+    void SendStartMetric();
+    // update app_info.json 
+    void UpdateAppInfoJson(const PipelineEventGroup& pipelineEventGroup);
+
     static const std::string INTERNAL_DATA_TYPE_ALARM;
     static const std::string INTERNAL_DATA_TYPE_METRIC;
 
@@ -51,7 +56,7 @@ class SelfMonitorServer {
     std::condition_variable mStopCV;
 
     // metrics
-    void SendMetrics();
+    void SendMetrics(bool updateAppInfo = false);
     bool ProcessSelfMonitorMetricEvent(SelfMonitorMetricEvent& event, const SelfMonitorMetricRule& rule);
     void PushSelfMonitorMetricEvents(std::vector<SelfMonitorMetricEvent>& events);
     void ReadAsPipelineEventGroup(PipelineEventGroup& pipelineEventGroup);