feat: prom agent info for HPA and Rebalance

core/monitor/Monitor.cpp

@@ -177,6 +177,8 @@ void LogtailMonitor::Monitor() {
 
                 GetMemStat();
                 CalCpuStat(curCpuStat, mCpuStat);
+                mCpuUsage.store(mCpuStat.mCpuUsage);
+                mMemoryUsage.store(mMemStat.mRss);
                 if (CheckHardMemLimit()) {
                     LOG_ERROR(sLogger,
                               ("Resource used by program exceeds hard limit",

core/monitor/Monitor.h

@@ -91,6 +91,8 @@ class LogtailMonitor {
     // GetRealtimeCpuLevel return a value to indicates current CPU usage level.
     // LogInput use it to do flow control.
     float GetRealtimeCpuLevel() { return mRealtimeCpuStat.mCpuUsage / mScaledCpuUsageUpLimit; }
+    [[nodiscard]] float GetCpuUsage() const { return mCpuUsage.load(); }
+    [[nodiscard]] float GetMemoryUsage() const { return mMemoryUsage.load(); }
 
 private:
     LogtailMonitor();
@@ -160,6 +162,9 @@ class LogtailMonitor {
     // Memory usage statistics.
     MemStat mMemStat;
 
+    std::atomic<float> mCpuUsage = 0;
+    std::atomic<float> mMemoryUsage = 0;
+
     // Current scale up level, updated by CheckScaledCpuUsageUpLimit.
     float mScaledCpuUsageUpLimit;
 #if defined(__linux__)

core/plugin/processor/inner/ProcessorPromRelabelMetricNative.cpp

@@ -16,8 +16,7 @@
 #include "plugin/processor/inner/ProcessorPromRelabelMetricNative.h"
 
 #include <cstddef>
-
-#include "json/json.h"
+#include <json/json.h>
 
 #include "common/Flags.h"
 #include "common/StringTools.h"
@@ -186,18 +185,26 @@ void ProcessorPromRelabelMetricNative::UpdateAutoMetrics(const PipelineEventGrou
 
 void ProcessorPromRelabelMetricNative::AddAutoMetrics(PipelineEventGroup& eGroup,
                                                       const prom::AutoMetric& autoMetric) const {
-    auto targetTags = eGroup.GetTags();
     if (!eGroup.HasMetadata(EventGroupMetaKey::PROMETHEUS_SCRAPE_TIMESTAMP_MILLISEC)) {
         LOG_ERROR(sLogger, ("scrape_timestamp_milliseconds is not set", ""));
         return;
     }
+    auto targetTags = eGroup.GetTags();
+    auto toDelete = GetToDeleteTargetLabels(targetTags);
+    for (const auto& item : toDelete) {
+        targetTags.erase(item);
+    }
+    if (!eGroup.HasMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_ID)) {
+        LOG_ERROR(sLogger, ("prometheus stream id", ""));
+        return;
+    }
+    targetTags[prometheus::LC_TARGET_HASH] = eGroup.GetMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_ID);
 
     StringView scrapeTimestampMilliSecStr = eGroup.GetMetadata(EventGroupMetaKey::PROMETHEUS_SCRAPE_TIMESTAMP_MILLISEC);
     auto timestampMilliSec = StringTo<uint64_t>(scrapeTimestampMilliSecStr.to_string());
     auto timestamp = timestampMilliSec / 1000;
     auto nanoSec = timestampMilliSec % 1000 * 1000000;
 
-
     AddMetric(
         eGroup, prometheus::SCRAPE_DURATION_SECONDS, autoMetric.mScrapeDurationSeconds, timestamp, nanoSec, targetTags);
 

core/prometheus/Constants.h

@@ -13,6 +13,7 @@ const uint64_t RefeshIntervalSeconds = 5;
 const char* const META = "__meta_";
 const char* const UNDEFINED = "undefined";
 const std::string PROMETHEUS = "prometheus";
+const char* const LC_TARGET_HASH = "lc_target_hash";
 
 // relabel config
 const char* const SOURCE_LABELS = "source_labels";
@@ -100,6 +101,8 @@ const char* const SCHEME_LABEL_NAME = "__scheme__";
 const char* const METRICS_PATH_LABEL_NAME = "__metrics_path__";
 const char* const PARAM_LABEL_NAME = "__param_";
 const char* const LABELS = "labels";
+const char* const TARGET_HASH = "hash";
+const char* const REBALANCE_MS = "rebalance_ms";
 
 // auto metrics
 const char* const SCRAPE_STATE = "scrape_state";
@@ -119,4 +122,14 @@ const char* const ACCEPT_ENCODING = "Accept-Encoding";
 const char* const GZIP = "gzip";
 const char* const IDENTITY = "identity";
 
+const char* const AGENT_INFO = "AgentInfo";
+const char* const TARGETS_INFO = "TargetsInfo";
+const char* const CPU_LIMIT = "CpuLimit";
+const char* const CPU_USAGE = "CpuUsage";
+const char* const MEM_LIMIT = "MemLimit";
+const char* const MEM_USAGE = "MemUsage";
+const char* const HASH = "Hash";
+const char* const SIZE = "Size";
+const char* const SCRAPE_DELAY_SECONDS = "ScrapeDelaySeconds";
+
 } // namespace logtail::prometheus

core/prometheus/PrometheusInputRunner.cpp

@@ -22,6 +22,7 @@
 #include <string>
 
 #include "application/Application.h"
+#include "collection_pipeline/queue/ProcessQueueManager.h"
 #include "common/Flags.h"
 #include "common/JsonUtil.h"
 #include "common/StringTools.h"
@@ -31,6 +32,7 @@
 #include "common/http/Curl.h"
 #include "common/timer/Timer.h"
 #include "logger/Logger.h"
+#include "monitor/Monitor.h"
 #include "monitor/metric_constants/MetricConstants.h"
 #include "plugin/flusher/sls/FlusherSLS.h"
 #include "prometheus/Constants.h"
@@ -51,6 +53,7 @@ PrometheusInputRunner::PrometheusInputRunner()
       mEventPool(true),
       mUnRegisterMs(0) {
     mTimer = std::make_shared<Timer>();
+    mLastUpdateTime = std::chrono::steady_clock::now();
 
     // self monitor
     MetricLabels labels;
@@ -175,6 +178,8 @@ void PrometheusInputRunner::Init() {
                                 mUnRegisterMs = 0;
                             } else {
                                 mUnRegisterMs.store(StringTo<uint64_t>(tmpStr));
+                                // adjust unRegisterMs to scrape targets for zero-cost
+                                mUnRegisterMs -= 1000;
                                 LOG_INFO(sLogger, ("unRegisterMs", ToString(mUnRegisterMs)));
                             }
                         }
@@ -293,4 +298,34 @@ string PrometheusInputRunner::GetAllProjects() {
 void PrometheusInputRunner::CheckGC() {
     mEventPool.CheckGC();
 }
+
+PromAgentInfo PrometheusInputRunner::GetAgentInfo() {
+    std::lock_guard<mutex> lock(mAgentInfoMutex);
+    auto curTime = std::chrono::steady_clock::now();
+#ifdef APSARA_UNIT_TEST_MAIN
+    curTime += std::chrono::seconds(prometheus::RefeshIntervalSeconds);
+#endif
+    if (curTime - mLastUpdateTime >= std::chrono::seconds(prometheus::RefeshIntervalSeconds)) {
+        mLastUpdateTime = curTime;
+        mAgentInfo.mCpuUsage = LogtailMonitor::GetInstance()->GetCpuUsage();
+        mAgentInfo.mMemUsage = LogtailMonitor::GetInstance()->GetMemoryUsage();
+        mAgentInfo.mCpuLimit = AppConfig::GetInstance()->GetCpuUsageUpLimit();
+        mAgentInfo.mMemLimit = AppConfig::GetInstance()->GetMemUsageUpLimit();
+
+        int queueNums = 0;
+        int validToPushNums = 0;
+
+        {
+            ReadLock lock(mSubscriberMapRWLock);
+            queueNums = mTargetSubscriberSchedulerMap.size();
+            for (auto& [k, v] : mTargetSubscriberSchedulerMap) {
+                if (ProcessQueueManager::GetInstance()->IsValidToPush(v->mQueueKey)) {
+                    validToPushNums++;
+                }
+            }
+        }
+    }
+
+    return mAgentInfo;
+}
 }; // namespace logtail

core/prometheus/PrometheusInputRunner.h

@@ -30,6 +30,13 @@
 
 namespace logtail {
 
+struct PromAgentInfo {
+    float mCpuUsage;
+    float mMemUsage;
+    float mCpuLimit;
+    float mMemLimit;
+};
+
 class PrometheusInputRunner : public InputRunner {
 public:
     PrometheusInputRunner(const PrometheusInputRunner&) = delete;
@@ -42,6 +49,7 @@ class PrometheusInputRunner : public InputRunner {
         return &sInstance;
     }
     void CheckGC();
+    PromAgentInfo GetAgentInfo();
 
     // input plugin update
     void UpdateScrapeInput(std::shared_ptr<TargetSubscriberScheduler> targetSubscriber,
@@ -74,6 +82,10 @@ class PrometheusInputRunner : public InputRunner {
     int32_t mServicePort;
     std::string mPodName;
 
+    std::mutex mAgentInfoMutex;
+    PromAgentInfo mAgentInfo{0.0F, 0.0F, 0.0F, 0.0F};
+    std::chrono::steady_clock::time_point mLastUpdateTime;
+
     std::shared_ptr<Timer> mTimer;
     EventPool mEventPool;
 

core/prometheus/component/StreamScraper.cpp

@@ -102,7 +102,7 @@ void StreamScraper::PushEventGroup(PipelineEventGroup&& eGroup) const {
 void StreamScraper::SendMetrics() {
     mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_SCRAPE_TIMESTAMP_MILLISEC,
                             ToString(mScrapeTimestampMilliSec));
-    mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_ID, GetId() + ToString(mScrapeTimestampMilliSec));
+    mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_ID, GetId());
 
     SetTargetLabels(mEventGroup);
     PushEventGroup(std::move(mEventGroup));
@@ -127,7 +127,7 @@ void StreamScraper::SetAutoMetricMeta(double scrapeDurationSeconds, bool upState
     mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_SCRAPE_DURATION, ToString(scrapeDurationSeconds));
     mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_SCRAPE_RESPONSE_SIZE, ToString(mRawSize));
     mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_UP_STATE, ToString(upState));
-    mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_ID, GetId() + ToString(mScrapeTimestampMilliSec));
+    mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_ID, GetId());
     mEventGroup.SetMetadata(EventGroupMetaKey::PROMETHEUS_STREAM_TOTAL, ToString(mStreamIndex));
 }
 std::string StreamScraper::GetId() {

core/prometheus/labels/Labels.cpp

@@ -111,10 +111,26 @@ void Labels::Range(const std::function<void(const string& k, const string& v)>&
 uint64_t Labels::Hash() {
     string hash;
     uint64_t sum = prometheus::OFFSET64;
-    Range([&hash](const string& k, const string& v) { hash += k + "\xff" + v + "\xff"; });
-    for (auto i : hash) {
-        sum ^= (uint64_t)i;
-        sum *= prometheus::PRIME64;
+    vector<string> names;
+    Range([&names](const string& k, const string&) { names.push_back(k); });
+    sort(names.begin(), names.end());
+    auto calc = [](uint64_t h, uint64_t c) {
+        h ^= (uint64_t)c;
+        h *= prometheus::PRIME64;
+        return h;
+    };
+    auto calcString = [](uint64_t h, const string& s) {
+        for (auto c : s) {
+            h ^= (uint64_t)c;
+            h *= prometheus::PRIME64;
+        }
+        return h;
+    };
+    for (const auto& name : names) {
+        sum = calcString(sum, name);
+        sum = calc(sum, 255);
+        sum = calcString(sum, Get(name));
+        sum = calc(sum, 255);
     }
     return sum;
 }

core/prometheus/schedulers/BaseScheduler.cpp

@@ -8,6 +8,10 @@ using namespace std;
 namespace logtail {
 void BaseScheduler::ExecDone() {
     mExecCount++;
+    while (mLatestExecTime > mFirstExecTime + chrono::seconds(mExecCount * mInterval)) {
+        mExecCount++;
+    }
+
     mLatestExecTime = mFirstExecTime + chrono::seconds(mExecCount * mInterval);
     mLatestScrapeTime = mFirstScrapeTime + chrono::seconds(mExecCount * mInterval);
 }

core/prometheus/schedulers/ScrapeScheduler.cpp

@@ -40,22 +40,18 @@ using namespace std;
 namespace logtail {
 
 ScrapeScheduler::ScrapeScheduler(std::shared_ptr<ScrapeConfig> scrapeConfigPtr,
-                                 std::string host,
+                                 string host,
                                  int32_t port,
-                                 Labels labels,
                                  QueueKey queueKey,
-                                 size_t inputIndex)
+                                 size_t inputIndex,
+                                 const PromTargetInfo& targetInfo)
     : mScrapeConfigPtr(std::move(scrapeConfigPtr)),
       mHost(std::move(host)),
       mPort(port),
+      mTargetInfo(targetInfo),
       mQueueKey(queueKey),
       mInputIndex(inputIndex),
-      mTargetLabels(labels) {
-    string tmpTargetURL = mScrapeConfigPtr->mScheme + "://" + mHost + ":" + ToString(mPort)
-        + mScrapeConfigPtr->mMetricsPath
-        + (mScrapeConfigPtr->mQueryString.empty() ? "" : "?" + mScrapeConfigPtr->mQueryString);
-    mHash = mScrapeConfigPtr->mJobName + tmpTargetURL + ToString(labels.Hash());
-    mInstance = mHost + ":" + ToString(mPort);
+      mScrapeResponseSizeBytes(-1) {
     mInterval = mScrapeConfigPtr->mScrapeIntervalSeconds;
 }
 
@@ -83,28 +79,27 @@ void ScrapeScheduler::OnMetricResult(HttpResponse& response, uint64_t) {
         scrapeState = prom::NetworkCodeToState(NetworkCode::Ok);
     }
 
-    mScrapeDurationSeconds = scrapeDurationMilliSeconds * sRate;
-    mUpState = response.GetStatusCode() == 200;
     if (response.GetStatusCode() != 200) {
         LOG_WARNING(sLogger,
-                    ("scrape failed, status code",
-                     response.GetStatusCode())("target", mHash)("curl msg", response.GetNetworkStatus().mMessage));
+                    ("scrape failed, status code", response.GetStatusCode())("target", mTargetInfo.mHash)(
+                        "curl msg", response.GetNetworkStatus().mMessage));
     }
 
-    auto mScrapeDurationSeconds = scrapeDurationMilliSeconds * sRate;
-    auto mUpState = response.GetStatusCode() == 200;
+    auto scrapeDurationSeconds = scrapeDurationMilliSeconds * sRate;
+    auto upState = response.GetStatusCode() == 200;
     streamScraper->mStreamIndex++;
     streamScraper->FlushCache();
-    streamScraper->SetAutoMetricMeta(mScrapeDurationSeconds, mUpState, scrapeState);
+    streamScraper->SetAutoMetricMeta(scrapeDurationSeconds, upState, scrapeState);
     streamScraper->SendMetrics();
+    mScrapeResponseSizeBytes = streamScraper->mRawSize;
     streamScraper->Reset();
 
     mPluginTotalDelayMs->Add(scrapeDurationMilliSeconds);
 }
 
 
 string ScrapeScheduler::GetId() const {
-    return mHash;
+    return mTargetInfo.mHash;
 }
 
 void ScrapeScheduler::SetComponent(shared_ptr<Timer> timer, EventPool* eventPool) {
@@ -126,6 +121,7 @@ void ScrapeScheduler::ScheduleNext() {
             return true;
         }
         this->DelayExecTime(1);
+        this->mExecDelayCount++;
         this->mPromDelayTotal->Add(1);
         this->ScheduleNext();
         return false;
@@ -176,7 +172,8 @@ std::unique_ptr<TimerEvent> ScrapeScheduler::BuildScrapeTimerEvent(std::chrono::
         mScrapeConfigPtr->mRequestHeaders,
         "",
         HttpResponse(
-            new prom::StreamScraper(mTargetLabels, mQueueKey, mInputIndex, mHash, mEventPool, mLatestScrapeTime),
+            new prom::StreamScraper(
+                mTargetInfo.mLabels, mQueueKey, mInputIndex, mTargetInfo.mHash, mEventPool, mLatestScrapeTime),
             [](void* p) { delete static_cast<prom::StreamScraper*>(p); },
             prom::StreamScraper::MetricWriteCallback),
         mScrapeConfigPtr->mScrapeTimeoutSeconds,
@@ -206,7 +203,7 @@ void ScrapeScheduler::Cancel() {
 void ScrapeScheduler::InitSelfMonitor(const MetricLabels& defaultLabels) {
     mSelfMonitor = std::make_shared<PromSelfMonitorUnsafe>();
     MetricLabels labels = defaultLabels;
-    labels.emplace_back(METRIC_LABEL_KEY_INSTANCE, mInstance);
+    labels.emplace_back(METRIC_LABEL_KEY_INSTANCE, mTargetInfo.mInstance);
 
     static const std::unordered_map<std::string, MetricType> sScrapeMetricKeys
         = {{METRIC_PLUGIN_OUT_EVENTS_TOTAL, MetricType::METRIC_TYPE_COUNTER},