fix

core/common/timer/Timer.cpp

@@ -51,6 +51,13 @@ void Timer::Stop() {
     }
 }
 
+void Timer::Clear() {
+    lock_guard<mutex> lock(mQueueMux);
+    while (!mQueue.empty()) {
+        mQueue.pop();
+    }
+}
+
 void Timer::PushEvent(unique_ptr<TimerEvent>&& e) {
     lock_guard<mutex> lock(mQueueMux);
     if (mQueue.empty() || e->GetExecTime() < mQueue.top()->GetExecTime()) {

core/common/timer/Timer.h

@@ -46,6 +46,7 @@ class Timer {
     void Init();
     void Stop();
     void PushEvent(std::unique_ptr<TimerEvent>&& e);
+    void Clear();
 
 private:
     Timer() = default;
@@ -57,7 +58,7 @@ class Timer {
 
     std::future<void> mThreadRes;
     mutable std::mutex mThreadRunningMux;
-    bool mIsThreadRunning = true;
+    bool mIsThreadRunning = false;
     mutable std::condition_variable mCV;
 
 #ifdef APSARA_UNIT_TEST_MAIN

core/host_monitor/HostMonitorInputRunner.cpp

@@ -45,12 +45,17 @@ void HostMonitorInputRunner::UpdateCollector(const std::vector<std::string>& new
     for (auto& collector : newCollectors) {
         auto oldCollector = mRegisteredCollectorMap.find(collector);
         if (oldCollector == mRegisteredCollectorMap.end()) {
+            if (mCollectorInstanceMap.find(collector) == mCollectorInstanceMap.end()) {
+                LOG_ERROR(sLogger, ("host monitor", "collector not found")("collector", collector));
+                continue;
+            }
             mRegisteredCollectorMap[collector] = true;
             HostMonitorTimerEvent::CollectConfig collectConfig(
                 collector, processQueueKey, inputIndex, std::chrono::seconds(DEFAULT_SCHEDULE_INTERVAL));
             auto now = std::chrono::steady_clock::now();
             auto event = std::make_unique<HostMonitorTimerEvent>(now + collectConfig.mInterval, collectConfig);
             Timer::GetInstance()->PushEvent(std::move(event));
+            LOG_INFO(sLogger, ("host monitor", "add new collector")("collector", collector));
         } else {
             // config removed and added again, timer event is still in the queue
             if (!oldCollector->second) {
@@ -122,6 +127,7 @@ void HostMonitorInputRunner::ScheduleOnce(HostMonitorTimerEvent::CollectConfig&
             return;
         }
 
+        LOG_DEBUG(sLogger, ("host monitor collect", "collector")(config.mCollectorName, group.GetEvents().size()));
         if (group.GetEvents().size() > 0) {
             bool result = ProcessorRunner::GetInstance()->PushQueue(
                 config.mProcessQueueKey, config.mInputIndex, std::move(group));

core/plugin/input/InputHostMeta.cpp

@@ -30,7 +30,7 @@ bool InputHostMeta::Init(const Json::Value& config, Json::Value& optionalGoPipel
 bool InputHostMeta::Start() {
     LOG_INFO(sLogger, ("input host meta start", mContext->GetConfigName()));
     HostMonitorInputRunner::GetInstance()->Init();
-    HostMonitorInputRunner::GetInstance()->UpdateCollector({"process"}, mContext->GetProcessQueueKey(), mIndex);
+    HostMonitorInputRunner::GetInstance()->UpdateCollector({"process_entity"}, mContext->GetProcessQueueKey(), mIndex);
     return true;
 }
 

core/prometheus/PrometheusInputRunner.cpp

@@ -137,7 +137,6 @@ void PrometheusInputRunner::Init() {
     mIsStarted = true;
 
 #ifndef APSARA_UNIT_TEST_MAIN
-    mTimer->Init();
     AsynCurlRunner::GetInstance()->Init();
 #endif
 
@@ -203,7 +202,6 @@ void PrometheusInputRunner::Stop() {
     }
 
 #ifndef APSARA_UNIT_TEST_MAIN
-    mTimer->Stop();
     LOG_INFO(sLogger, ("PrometheusInputRunner", "stop asyn curl runner"));
     AsynCurlRunner::GetInstance()->Stop();
 #endif

core/prometheus/PrometheusInputRunner.h

@@ -75,7 +75,6 @@ class PrometheusInputRunner : public InputRunner {
     std::string mPodName;
 
     std::unique_ptr<sdk::CurlClient> mClient;
-    std::shared_ptr<Timer> mTimer;
     EventPool mEventPool;
 
     mutable ReadWriteLock mSubscriberMapRWLock;

core/prometheus/schedulers/BaseScheduler.cpp

@@ -40,7 +40,6 @@ bool BaseScheduler::IsCancelled() {
 }
 
 void BaseScheduler::SetComponent(EventPool* eventPool) {
-    mTimer = Timer::GetInstance();
     mEventPool = eventPool;
 }
 } // namespace logtail

core/prometheus/schedulers/BaseScheduler.h

@@ -20,7 +20,8 @@ class BaseScheduler {
 
     std::chrono::steady_clock::time_point GetNextExecTime();
 
-    void SetFirstExecTime(std::chrono::steady_clock::time_point firstExecTime,std::chrono::system_clock::time_point firstScrapeTime);
+    void SetFirstExecTime(std::chrono::steady_clock::time_point firstExecTime,
+                          std::chrono::system_clock::time_point firstScrapeTime);
     void DelayExecTime(uint64_t delaySeconds);
     virtual void Cancel();
 
@@ -44,7 +45,6 @@ class BaseScheduler {
     std::shared_ptr<PromFuture<HttpResponse&, uint64_t>> mFuture;
     std::shared_ptr<PromFuture<>> mIsContextValidFuture;
 
-    Timer* mTimer;
     EventPool* mEventPool = nullptr;
 };
 } // namespace logtail

core/prometheus/schedulers/ScrapeScheduler.cpp

@@ -190,7 +190,7 @@ void ScrapeScheduler::ScheduleNext() {
     }
 
     auto event = BuildScrapeTimerEvent(GetNextExecTime());
-    mTimer->PushEvent(std::move(event));
+    Timer::GetInstance()->PushEvent(std::move(event));
 }
 
 void ScrapeScheduler::ScrapeOnce(std::chrono::steady_clock::time_point execTime) {
@@ -201,9 +201,7 @@ void ScrapeScheduler::ScrapeOnce(std::chrono::steady_clock::time_point execTime)
     });
     mFuture = future;
     auto event = BuildScrapeTimerEvent(execTime);
-    if (mTimer) {
-        mTimer->PushEvent(std::move(event));
-    }
+    Timer::GetInstance()->PushEvent(std::move(event));
 }
 
 std::unique_ptr<TimerEvent> ScrapeScheduler::BuildScrapeTimerEvent(std::chrono::steady_clock::time_point execTime) {

core/prometheus/schedulers/TargetSubscriberScheduler.cpp

@@ -106,25 +106,23 @@ void TargetSubscriberScheduler::UpdateScrapeScheduler(
         for (const auto& [k, v] : newScrapeSchedulerMap) {
             if (mScrapeSchedulerMap.find(k) == mScrapeSchedulerMap.end()) {
                 mScrapeSchedulerMap[k] = v;
-                if (mTimer) {
-                    auto tmpCurrentMilliSeconds = GetCurrentTimeInMilliSeconds();
-                    auto tmpRandSleepMilliSec = GetRandSleepMilliSec(
-                        v->GetId(), mScrapeConfigPtr->mScrapeIntervalSeconds, tmpCurrentMilliSeconds);
-
-                    // zero-cost upgrade
-                    if (mUnRegisterMs > 0
-                        && (tmpCurrentMilliSeconds + tmpRandSleepMilliSec
-                                - (uint64_t)mScrapeConfigPtr->mScrapeIntervalSeconds * 1000
-                            > mUnRegisterMs)
-                        && (tmpCurrentMilliSeconds + tmpRandSleepMilliSec
-                                - (uint64_t)mScrapeConfigPtr->mScrapeIntervalSeconds * 1000 * 2
-                            < mUnRegisterMs)) {
-                        // scrape once just now
-                        LOG_INFO(sLogger, ("scrape zero cost", ToString(tmpCurrentMilliSeconds)));
-                        v->ScrapeOnce(std::chrono::steady_clock::now());
-                    }
-                    v->ScheduleNext();
+                auto tmpCurrentMilliSeconds = GetCurrentTimeInMilliSeconds();
+                auto tmpRandSleepMilliSec = GetRandSleepMilliSec(
+                    v->GetId(), mScrapeConfigPtr->mScrapeIntervalSeconds, tmpCurrentMilliSeconds);
+
+                // zero-cost upgrade
+                if (mUnRegisterMs > 0
+                    && (tmpCurrentMilliSeconds + tmpRandSleepMilliSec
+                            - (uint64_t)mScrapeConfigPtr->mScrapeIntervalSeconds * 1000
+                        > mUnRegisterMs)
+                    && (tmpCurrentMilliSeconds + tmpRandSleepMilliSec
+                            - (uint64_t)mScrapeConfigPtr->mScrapeIntervalSeconds * 1000 * 2
+                        < mUnRegisterMs)) {
+                    // scrape once just now
+                    LOG_INFO(sLogger, ("scrape zero cost", ToString(tmpCurrentMilliSeconds)));
+                    v->ScrapeOnce(std::chrono::steady_clock::now());
                 }
+                v->ScheduleNext();
             }
         }
     }
@@ -261,7 +259,7 @@ void TargetSubscriberScheduler::ScheduleNext() {
     }
 
     auto event = BuildSubscriberTimerEvent(GetNextExecTime());
-    mTimer->PushEvent(std::move(event));
+    Timer::GetInstance()->PushEvent(std::move(event));
 }
 
 void TargetSubscriberScheduler::Cancel() {
@@ -281,9 +279,7 @@ void TargetSubscriberScheduler::SubscribeOnce(std::chrono::steady_clock::time_po
     });
     mFuture = future;
     auto event = BuildSubscriberTimerEvent(execTime);
-    if (mTimer) {
-        mTimer->PushEvent(std::move(event));
-    }
+    Timer::GetInstance()->PushEvent(std::move(event));
 }
 
 std::unique_ptr<TimerEvent>

core/unittest/host_monitor/HostMonitorInputRunnerUnittest.cpp

@@ -33,6 +33,12 @@ class HostMonitorInputRunnerUnittest : public testing::Test {
 public:
     void TestUpdateAndRemoveCollector() const;
     void TestScheduleOnce() const;
+
+private:
+    void TearDown() override {
+        HostMonitorInputRunner::GetInstance()->Stop();
+        Timer::GetInstance()->Clear();
+    }
 };
 
 void HostMonitorInputRunnerUnittest::TestUpdateAndRemoveCollector() const {
@@ -67,8 +73,6 @@ void HostMonitorInputRunnerUnittest::TestScheduleOnce() const {
     APSARA_TEST_TRUE_FATAL(ProcessQueueManager::GetInstance()->PopItem(0, item, configName));
     APSARA_TEST_EQUAL_FATAL("test", configName);
 
-    // verify schdule next
-    APSARA_TEST_EQUAL_FATAL(Timer::GetInstance()->mQueue.size(), 1);
     runner->mThreadPool.Stop();
     runner->Stop();
 }

core/unittest/host_monitor/ProcessEntityCollectorUnittest.cpp

@@ -30,7 +30,7 @@ class ProcessEntityCollectorUnittest : public testing::Test {
 void ProcessEntityCollectorUnittest::TestGetNewProcessStat() const {
     PROCESS_DIR = ".";
     auto collector = ProcessEntityCollector();
-    auto ptr = collector.GetNewProcessStat(1);
+    auto ptr = collector.ReadNewProcessStat(1);
     APSARA_TEST_NOT_EQUAL(nullptr, ptr);
     APSARA_TEST_EQUAL(1, ptr->pid);
     APSARA_TEST_EQUAL("cat", ptr->name);

core/unittest/prometheus/ScrapeSchedulerUnittest.cpp

@@ -54,6 +54,8 @@ class ScrapeSchedulerUnittest : public testing::Test {
         mScrapeConfig->mRequestHeaders = {{"Authorization", "Bearer xxxxx"}};
     }
 
+    void TearDown() override { Timer::GetInstance()->Clear(); }
+
 private:
     std::shared_ptr<ScrapeConfig> mScrapeConfig;
 };