feat: add input raw event and event pool

core/application/Application.cpp

@@ -49,6 +49,7 @@
 #include "pipeline/queue/SenderQueueManager.h"
 #include "plugin/flusher/sls/DiskBufferWriter.h"
 #include "plugin/input/InputFeedbackInterfaceRegistry.h"
+#include "prometheus/PrometheusInputRunner.h"
 #include "runner/FlusherRunner.h"
 #include "runner/ProcessorRunner.h"
 #include "runner/sink/http/HttpSink.h"
@@ -324,6 +325,7 @@ void Application::Start() { // GCOVR_EXCL_START
 
         // destruct event handlers here so that it will not block file reading task
         ConfigManager::GetInstance()->DeleteHandlers();
+        PrometheusInputRunner::GetInstance()->CheckGC();
 
         this_thread::sleep_for(chrono::seconds(1));
     }

core/plugin/processor/inner/ProcessorPromParseMetricNative.cpp

@@ -3,10 +3,10 @@
 #include <json/json.h>
 
 #include "common/StringTools.h"
-#include "models/LogEvent.h"
 #include "models/MetricEvent.h"
 #include "models/PipelineEventGroup.h"
 #include "models/PipelineEventPtr.h"
+#include "models/RawEvent.h"
 #include "prometheus/Constants.h"
 
 using namespace std;
@@ -43,7 +43,7 @@ void ProcessorPromParseMetricNative::Process(PipelineEventGroup& eGroup) {
 }
 
 bool ProcessorPromParseMetricNative::IsSupportedEvent(const PipelineEventPtr& e) const {
-    return e.Is<LogEvent>();
+    return e.Is<RawEvent>();
 }
 
 bool ProcessorPromParseMetricNative::ProcessEvent(PipelineEventPtr& e,
@@ -53,9 +53,9 @@ bool ProcessorPromParseMetricNative::ProcessEvent(PipelineEventPtr& e,
     if (!IsSupportedEvent(e)) {
         return false;
     }
-    auto& sourceEvent = e.Cast<LogEvent>();
+    auto& sourceEvent = e.Cast<RawEvent>();
     std::unique_ptr<MetricEvent> metricEvent = eGroup.CreateMetricEvent(true);
-    if (parser.ParseLine(sourceEvent.GetContent(prometheus::PROMETHEUS), *metricEvent)) {
+    if (parser.ParseLine(sourceEvent.GetContent(), *metricEvent)) {
         metricEvent->SetTag(string(prometheus::NAME), metricEvent->GetName());
         newEvents.emplace_back(std::move(metricEvent), true, nullptr);
     }

core/prometheus/PrometheusInputRunner.cpp

@@ -47,6 +47,7 @@ PrometheusInputRunner::PrometheusInputRunner()
     : mServiceHost(STRING_FLAG(loong_collector_operator_service)),
       mServicePort(INT32_FLAG(loong_collector_operator_service_port)),
       mPodName(STRING_FLAG(_pod_name_)),
+      mEventPool(true),
       mUnRegisterMs(0) {
     mClient = std::make_unique<sdk::CurlClient>();
     mTimer = std::make_shared<Timer>();
@@ -83,7 +84,7 @@ void PrometheusInputRunner::UpdateScrapeInput(std::shared_ptr<TargetSubscriberSc
     targetSubscriber->InitSelfMonitor(defaultLabels);
 
     targetSubscriber->mUnRegisterMs = mUnRegisterMs.load();
-    targetSubscriber->SetTimer(mTimer);
+    targetSubscriber->SetComponent(mTimer, &mEventPool);
     auto randSleepMilliSec = GetRandSleepMilliSec(
         targetSubscriber->GetId(), prometheus::RefeshIntervalSeconds, GetCurrentTimeInMilliSeconds());
     auto firstExecTime = std::chrono::steady_clock::now() + std::chrono::milliseconds(randSleepMilliSec);
@@ -294,4 +295,8 @@ string PrometheusInputRunner::GetAllProjects() {
     }
     return result;
 }
+
+void PrometheusInputRunner::CheckGC() {
+    mEventPool.CheckGC();
+}
 }; // namespace logtail

core/prometheus/PrometheusInputRunner.h

@@ -22,7 +22,6 @@
 
 #include "common/Lock.h"
 #include "common/timer/Timer.h"
-#include "monitor/MetricManager.h"
 #include "monitor/MetricTypes.h"
 #include "prometheus/schedulers/TargetSubscriberScheduler.h"
 #include "runner/InputRunner.h"
@@ -42,6 +41,7 @@ class PrometheusInputRunner : public InputRunner {
         static PrometheusInputRunner sInstance;
         return &sInstance;
     }
+    void CheckGC();
 
     // input plugin update
     void UpdateScrapeInput(std::shared_ptr<TargetSubscriberScheduler> targetSubscriber,
@@ -70,13 +70,13 @@ class PrometheusInputRunner : public InputRunner {
     std::atomic<bool> mIsThreadRunning = true;
     std::future<void> mThreadRes;
 
-    std::unique_ptr<sdk::CurlClient> mClient;
-
     std::string mServiceHost;
     int32_t mServicePort;
     std::string mPodName;
 
+    std::unique_ptr<sdk::CurlClient> mClient;
     std::shared_ptr<Timer> mTimer;
+    EventPool mEventPool;
 
     mutable ReadWriteLock mSubscriberMapRWLock;
     std::map<std::string, std::shared_ptr<TargetSubscriberScheduler>> mTargetSubscriberSchedulerMap;

core/prometheus/schedulers/BaseScheduler.cpp

@@ -1,5 +1,10 @@
 #include "prometheus/schedulers/BaseScheduler.h"
 
+#include "common/timer/Timer.h"
+#include "models/EventPool.h"
+
+using namespace std;
+
 namespace logtail {
 void BaseScheduler::ExecDone() {
     mExecCount++;
@@ -28,4 +33,9 @@ bool BaseScheduler::IsCancelled() {
     ReadLock lock(mLock);
     return !mValidState;
 }
+
+void BaseScheduler::SetComponent(shared_ptr<Timer> timer, EventPool* eventPool) {
+    mTimer = std::move(timer);
+    mEventPool = eventPool;
+}
 } // namespace logtail

core/prometheus/schedulers/BaseScheduler.h

@@ -4,6 +4,8 @@
 #include <memory>
 
 #include "common/http/HttpResponse.h"
+#include "common/timer/Timer.h"
+#include "models/EventPool.h"
 #include "prometheus/async/PromFuture.h"
 
 namespace logtail {
@@ -20,9 +22,10 @@ class BaseScheduler {
 
     void SetFirstExecTime(std::chrono::steady_clock::time_point firstExecTime);
     void DelayExecTime(uint64_t delaySeconds);
-
     virtual void Cancel();
 
+    void SetComponent(std::shared_ptr<Timer> timer, EventPool* eventPool);
+
 protected:
     bool IsCancelled();
 
@@ -35,5 +38,8 @@ class BaseScheduler {
     bool mValidState = true;
     std::shared_ptr<PromFuture<HttpResponse&, uint64_t>> mFuture;
     std::shared_ptr<PromFuture<>> mIsContextValidFuture;
+
+    std::shared_ptr<Timer> mTimer;
+    EventPool* mEventPool = nullptr;
 };
 } // namespace logtail

core/prometheus/schedulers/ScrapeScheduler.cpp

@@ -24,13 +24,11 @@
 #include "common/StringTools.h"
 #include "common/TimeUtil.h"
 #include "common/timer/HttpRequestTimerEvent.h"
-#include "common/timer/Timer.h"
 #include "logger/Logger.h"
 #include "pipeline/queue/ProcessQueueItem.h"
 #include "pipeline/queue/ProcessQueueManager.h"
 #include "pipeline/queue/QueueKey.h"
 #include "prometheus/Constants.h"
-#include "prometheus/Utils.h"
 #include "prometheus/async/PromFuture.h"
 #include "prometheus/async/PromHttpRequest.h"
 #include "sdk/Common.h"
@@ -86,8 +84,6 @@ ScrapeScheduler::ScrapeScheduler(std::shared_ptr<ScrapeConfig> scrapeConfigPtr,
     mHash = mScrapeConfigPtr->mJobName + tmpTargetURL + ToString(mTargetLabels.Hash());
     mInstance = mHost + ":" + ToString(mPort);
     mInterval = mScrapeConfigPtr->mScrapeIntervalSeconds;
-
-    mParser = make_unique<TextParser>();
 }
 
 void ScrapeScheduler::OnMetricResult(HttpResponse& response, uint64_t timestampMilliSec) {
@@ -214,7 +210,7 @@ std::unique_ptr<TimerEvent> ScrapeScheduler::BuildScrapeTimerEvent(std::chrono::
                                             mScrapeConfigPtr->mRequestHeaders,
                                             "",
                                             HttpResponse(
-                                                new PromMetricResponseBody(),
+                                                new PromMetricResponseBody(mEventPool),
                                                 [](void* ptr) { delete static_cast<PromMetricResponseBody*>(ptr); },
                                                 PromMetricWriteCallback),
                                             mScrapeConfigPtr->mScrapeTimeoutSeconds,
@@ -238,10 +234,6 @@ void ScrapeScheduler::Cancel() {
     }
 }
 
-void ScrapeScheduler::SetTimer(std::shared_ptr<Timer> timer) {
-    mTimer = std::move(timer);
-}
-
 void ScrapeScheduler::InitSelfMonitor(const MetricLabels& defaultLabels) {
     mSelfMonitor = std::make_shared<PromSelfMonitorUnsafe>();
     MetricLabels labels = defaultLabels;

core/prometheus/schedulers/ScrapeScheduler.h

@@ -21,14 +21,11 @@
 
 #include "BaseScheduler.h"
 #include "common/http/HttpResponse.h"
-#include "common/timer/Timer.h"
 #include "models/PipelineEventGroup.h"
 #include "monitor/MetricTypes.h"
 #include "pipeline/queue/QueueKey.h"
-#include "prometheus/Constants.h"
 #include "prometheus/PromSelfMonitor.h"
 #include "prometheus/Utils.h"
-#include "prometheus/labels/TextParser.h"
 #include "prometheus/schedulers/ScrapeConfig.h"
 
 #ifdef APSARA_UNIT_TEST_MAIN
@@ -43,13 +40,15 @@ struct PromMetricResponseBody {
     PipelineEventGroup mEventGroup;
     std::string mCache;
     size_t mRawSize = 0;
+    EventPool* mEventPool = nullptr;
 
-    PromMetricResponseBody() : mEventGroup(std::make_shared<SourceBuffer>()) {};
+    explicit PromMetricResponseBody(EventPool* eventPool)
+        : mEventGroup(std::make_shared<SourceBuffer>()), mEventPool(eventPool) {};
     void AddEvent(char* line, size_t len) {
         if (IsValidMetric(StringView(line, len))) {
-            auto* e = mEventGroup.AddLogEvent();
+            auto* e = mEventGroup.AddRawEvent(true, mEventPool);
             auto sb = mEventGroup.GetSourceBuffer()->CopyString(line, len);
-            e->SetContentNoCopy(prometheus::PROMETHEUS, StringView(sb.data, sb.size));
+            e->SetContentNoCopy(sb);
         }
     }
     void FlushCache() {
@@ -70,7 +69,6 @@ class ScrapeScheduler : public BaseScheduler {
     ~ScrapeScheduler() override = default;
 
     void OnMetricResult(HttpResponse&, uint64_t timestampMilliSec);
-    void SetTimer(std::shared_ptr<Timer> timer);
 
     std::string GetId() const;
 
@@ -94,11 +92,8 @@ class ScrapeScheduler : public BaseScheduler {
     std::string mInstance;
     Labels mTargetLabels;
 
-    std::unique_ptr<TextParser> mParser;
-
     QueueKey mQueueKey;
     size_t mInputIndex;
-    std::shared_ptr<Timer> mTimer;
 
     // auto metrics
     uint64_t mScrapeTimestampMilliSec = 0;

core/prometheus/schedulers/TargetSubscriberScheduler.cpp

@@ -223,7 +223,7 @@ TargetSubscriberScheduler::BuildScrapeSchedulerSet(std::vector<Labels>& targetGr
         auto scrapeScheduler
             = std::make_shared<ScrapeScheduler>(mScrapeConfigPtr, host, port, resultLabel, mQueueKey, mInputIndex);
 
-        scrapeScheduler->SetTimer(mTimer);
+        scrapeScheduler->SetComponent(mTimer, mEventPool);
 
         auto randSleepMilliSec = GetRandSleepMilliSec(
             scrapeScheduler->GetId(), mScrapeConfigPtr->mScrapeIntervalSeconds, GetCurrentTimeInMilliSeconds());
@@ -236,9 +236,6 @@ TargetSubscriberScheduler::BuildScrapeSchedulerSet(std::vector<Labels>& targetGr
     return scrapeSchedulerMap;
 }
 
-void TargetSubscriberScheduler::SetTimer(shared_ptr<Timer> timer) {
-    mTimer = std::move(timer);
-}
 
 string TargetSubscriberScheduler::GetId() const {
     return mJobName;

core/prometheus/schedulers/TargetSubscriberScheduler.h

@@ -41,7 +41,6 @@ class TargetSubscriberScheduler : public BaseScheduler {
     bool operator<(const TargetSubscriberScheduler& other) const;
 
     void OnSubscription(HttpResponse&, uint64_t);
-    void SetTimer(std::shared_ptr<Timer> timer);
     void SubscribeOnce(std::chrono::steady_clock::time_point execTime);
 
     std::string GetId() const;
@@ -79,7 +78,6 @@ class TargetSubscriberScheduler : public BaseScheduler {
     std::unordered_map<std::string, std::shared_ptr<ScrapeScheduler>> mScrapeSchedulerMap;
 
     std::string mJobName;
-    std::shared_ptr<Timer> mTimer;
 
     std::string mETag;
 

core/unittest/processor/ProcessorPromParseMetricNativeUnittest.cpp

@@ -85,8 +85,8 @@ void ProcessorParsePrometheusMetricUnittest::TestProcess() {
             if (newLine.empty() || newLine[0] == '#') {
                 continue;
             }
-            auto* MetricEvent = eGroup.AddLogEvent();
-            MetricEvent->SetContent(prometheus::PROMETHEUS, newLine);
+            auto* metricEvent = eGroup.AddRawEvent();
+            metricEvent->SetContent(newLine);
         }
 
         return eGroup;