[2650-w0] feat(file_server): add WatchManager runner skeleton (pass-through)

core/file_server/EventDispatcher.cpp

@@ -39,6 +39,7 @@
 #include "common/StringTools.h"
 #include "common/TimeUtil.h"
 #include "common/version.h"
+#include "file_server/WatchManager.h"
 #include "file_server/checkpoint/CheckPointManager.h"
 #include "file_server/checkpoint/CheckpointManagerV2.h"
 #include "file_server/event/Event.h"
@@ -239,7 +240,7 @@ bool EventDispatcher::RegisterEventHandler(const string& path,
     } else {
         // need check mEventListener valid
         if (mEventListener->IsInit() && !AppConfig::GetInstance()->IsInInotifyBlackList(path)) {
-            wd = mEventListener->AddWatch(path.c_str());
+            wd = WatchManager::GetInstance()->AcquireWatch(path);
             if (!EventListener::IsValidID(wd)) {
                 string str = ErrnoToString(GetErrno());
                 LOG_WARNING(sLogger, ("failed to register dir", path)("reason", str));
@@ -868,7 +869,7 @@ void EventDispatcher::UnregisterEventHandler(const string& path) {
     RemoveOneToOneMapEntry(wd);
     mWdUpdateTimeMap.erase(wd);
     if (EventListener::IsValidID(wd) && mEventListener->IsInit()) {
-        mEventListener->RemoveWatch(wd);
+        WatchManager::GetInstance()->ReleaseWatch(wd);
         mInotifyWatchNum--;
     }
     mWatchNum--;

core/file_server/WatchManager.cpp

@@ -0,0 +1,45 @@
+// Copyright 2025 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "file_server/WatchManager.h"
+
+using namespace std;
+
+namespace logtail {
+
+WatchManager::WatchManager() : mEventListener(EventListener::GetInstance()) {
+    WriteMetrics::GetInstance()->CreateMetricsRecordRef(
+        mMetricsRecordRef,
+        MetricCategory::METRIC_CATEGORY_RUNNER,
+        {{METRIC_LABEL_KEY_RUNNER_NAME, METRIC_LABEL_VALUE_RUNNER_NAME_WATCH_MANAGER}});
+    WriteMetrics::GetInstance()->CommitMetricsRecordRef(mMetricsRecordRef);
+}
+
+int WatchManager::AcquireWatch(const string& path) {
+    return DoAddWatch(path.c_str());
+}
+
+bool WatchManager::ReleaseWatch(int wd) {
+    return DoRemoveWatch(wd);
+}
+
+int WatchManager::DoAddWatch(const char* path) {
+    return mEventListener->AddWatch(path);
+}
+
+bool WatchManager::DoRemoveWatch(int wd) {
+    return mEventListener->RemoveWatch(wd);
+}
+
+} // namespace logtail

core/file_server/WatchManager.h

@@ -0,0 +1,76 @@
+// Copyright 2025 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <string>
+
+#include "file_server/event_listener/EventListener.h"
+#include "monitor/MetricManager.h"
+#include "monitor/metric_constants/MetricConstants.h"
+#include "monitor/metric_models/MetricRecord.h"
+
+namespace logtail {
+
+// WatchManager is a dedicated runner that centralizes access to the underlying
+// EventListener (inotify on Linux). Callers acquire and release directory
+// watches through it instead of touching EventListener directly, which gives us
+// a single place to observe and, later, to manage watch lifecycle.
+//
+// This is the W0 skeleton: AcquireWatch/ReleaseWatch forward one-to-one to
+// EventListener::AddWatch/RemoveWatch with no added dedup, ref-counting or
+// lifecycle change (pass-through). Only a runner_name = "watch_manager"
+// MetricsRecordRef is registered to establish an observability baseline.
+class WatchManager {
+public:
+    static WatchManager* GetInstance() {
+        static WatchManager* ptr = new WatchManager();
+        return ptr;
+    }
+
+    virtual ~WatchManager() = default;
+
+    WatchManager(const WatchManager&) = delete;
+    WatchManager& operator=(const WatchManager&) = delete;
+
+    // Register a watch on path. Returns the underlying watch descriptor
+    // (identical to what EventListener::AddWatch would return, including the
+    // invalid id on failure).
+    int AcquireWatch(const std::string& path);
+
+    // Remove the watch identified by wd. Returns whether the underlying
+    // EventListener::RemoveWatch succeeded.
+    bool ReleaseWatch(int wd);
+
+    MetricsRecordRef& GetMetricsRecordRef() { return mMetricsRecordRef; }
+
+protected:
+    WatchManager();
+
+    // Underlying watch syscalls, forwarded to EventListener by default. Kept
+    // virtual so tests can substitute a counting fake and assert pass-through
+    // equivalence without depending on a live inotify instance.
+    virtual int DoAddWatch(const char* path);
+    virtual bool DoRemoveWatch(int wd);
+
+private:
+    EventListener* mEventListener = nullptr;
+    mutable MetricsRecordRef mMetricsRecordRef;
+
+#ifdef APSARA_UNIT_TEST_MAIN
+    friend class WatchManagerUnittest;
+#endif
+};
+
+} // namespace logtail

core/monitor/metric_constants/MetricConstants.h

@@ -300,6 +300,7 @@ extern const std::string METRIC_LABEL_VALUE_RUNNER_NAME_PROMETHEUS;
 extern const std::string METRIC_LABEL_VALUE_RUNNER_NAME_EBPF_SERVER;
 extern const std::string METRIC_LABEL_VALUE_RUNNER_NAME_K8S_METADATA;
 extern const std::string METRIC_LABEL_VALUE_RUNNER_NAME_STATIC_FILE_SERVER;
+extern const std::string METRIC_LABEL_VALUE_RUNNER_NAME_WATCH_MANAGER;
 
 // metric keys
 extern const std::string& METRIC_RUNNER_IN_EVENTS_TOTAL;

core/monitor/metric_constants/RunnerMetrics.cpp

@@ -32,6 +32,7 @@ const string METRIC_LABEL_VALUE_RUNNER_NAME_PROMETHEUS = "prometheus_runner";
 const string METRIC_LABEL_VALUE_RUNNER_NAME_EBPF_SERVER = "ebpf_runner";
 const string METRIC_LABEL_VALUE_RUNNER_NAME_K8S_METADATA = "k8s_metadata_runner";
 const string METRIC_LABEL_VALUE_RUNNER_NAME_STATIC_FILE_SERVER = "static_file_server";
+const string METRIC_LABEL_VALUE_RUNNER_NAME_WATCH_MANAGER = "watch_manager";
 
 // metric keys
 const string& METRIC_RUNNER_IN_EVENTS_TOTAL = METRIC_IN_EVENTS_TOTAL;

core/unittest/CMakeLists.txt

@@ -45,6 +45,7 @@ macro(add_core_subdir)
     add_subdirectory(controller)
     add_subdirectory(event)
     add_subdirectory(event_handler)
+    add_subdirectory(file_server)
     add_subdirectory(file_source)
     add_subdirectory(flusher)
     add_subdirectory(input)

core/unittest/file_server/CMakeLists.txt

@@ -0,0 +1,22 @@
+# Copyright 2025 iLogtail Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+cmake_minimum_required(VERSION 3.22)
+project(file_server_unittest)
+
+add_executable(watch_manager_unittest WatchManagerUnittest.cpp)
+target_link_libraries(watch_manager_unittest ${UT_BASE_TARGET})
+
+include(GoogleTest)
+gtest_discover_tests(watch_manager_unittest)

core/unittest/file_server/WatchManagerUnittest.cpp

@@ -0,0 +1,149 @@
+// Copyright 2025 iLogtail Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <string>
+#include <vector>
+
+#include "file_server/WatchManager.h"
+#include "file_server/event_listener/EventListener.h"
+#include "monitor/MetricManager.h"
+#include "monitor/metric_constants/MetricConstants.h"
+#include "unittest/Unittest.h"
+
+using namespace std;
+
+namespace logtail {
+
+// Records every underlying watch call so the test can assert WatchManager
+// forwards one-to-one to EventListener without altering counts or return
+// values (pass-through equivalence).
+class CountingWatchManager : public WatchManager {
+public:
+    CountingWatchManager() : WatchManager() {}
+
+    int mAddCount = 0;
+    int mRemoveCount = 0;
+    std::vector<std::string> mAddedPaths;
+    std::vector<int> mRemovedWds;
+    int mNextWd = 100;
+
+protected:
+    int DoAddWatch(const char* path) override {
+        ++mAddCount;
+        mAddedPaths.emplace_back(path);
+        return mNextWd++;
+    }
+    bool DoRemoveWatch(int wd) override {
+        ++mRemoveCount;
+        mRemovedWds.push_back(wd);
+        return true;
+    }
+};
+
+class WatchManagerUnittest : public ::testing::Test {
+public:
+    // AcquireWatch/ReleaseWatch must forward to the underlying watch syscalls
+    // exactly once per call and return their result unchanged.
+    void TestAcquireReleasePassThrough() {
+        CountingWatchManager wm;
+
+        int wd0 = wm.AcquireWatch("/tmp/watch_manager_ut/a");
+        APSARA_TEST_EQUAL_FATAL(1, wm.mAddCount);
+        APSARA_TEST_EQUAL_FATAL(string("/tmp/watch_manager_ut/a"), wm.mAddedPaths[0]);
+        APSARA_TEST_EQUAL_FATAL(100, wd0); // returns underlying wd verbatim
+
+        int wd1 = wm.AcquireWatch("/tmp/watch_manager_ut/b");
+        APSARA_TEST_EQUAL_FATAL(2, wm.mAddCount);
+        APSARA_TEST_EQUAL_FATAL(101, wd1);
+
+        bool released = wm.ReleaseWatch(wd0);
+        APSARA_TEST_TRUE_FATAL(released);
+        APSARA_TEST_EQUAL_FATAL(1, wm.mRemoveCount);
+        APSARA_TEST_EQUAL_FATAL(wd0, wm.mRemovedWds[0]);
+    }
+
+    // The skeleton adds no dedup: adding the same path twice forwards twice,
+    // preserving the pre-refactor behavior where EventDispatcher owns dedup.
+    void TestNoDedupAtWatchManager() {
+        CountingWatchManager wm;
+
+        wm.AcquireWatch("/tmp/watch_manager_ut/same");
+        wm.AcquireWatch("/tmp/watch_manager_ut/same");
+        APSARA_TEST_EQUAL_FATAL(2, wm.mAddCount);
+        APSARA_TEST_EQUAL_FATAL(wm.mAddedPaths[0], wm.mAddedPaths[1]);
+    }
+
+    // WatchManager must forward watches to the very same EventListener that
+    // EventDispatcher reads and removes events through; otherwise a watch
+    // acquired via WatchManager would be invisible to the dispatcher's event
+    // loop. Both sides obtain the listener from EventListener::GetInstance() --
+    // a Meyers singleton (function-local static, private ctor) that yields one
+    // process-wide instance. Lock that invariant here so a future refactor
+    // giving WatchManager its own listener fails loudly instead of silently
+    // desyncing the two.
+    void TestSharedEventListenerInstance() {
+        WatchManager* wm = WatchManager::GetInstance();
+        // GetInstance() is stable across calls (same singleton every time)...
+        APSARA_TEST_EQUAL_FATAL(EventListener::GetInstance(), EventListener::GetInstance());
+        // ...and the runner stores exactly that canonical instance, which is the
+        // same one EventDispatcher's ctor captures via EventListener::GetInstance().
+        APSARA_TEST_EQUAL_FATAL(EventListener::GetInstance(), wm->mEventListener);
+    }
+
+    // Constructing WatchManager registers a runner_name = "watch_manager"
+    // MetricsRecordRef and commits it to WriteMetrics.
+    void TestMetricsRecordRegistered() {
+        WatchManager* wm = WatchManager::GetInstance();
+        APSARA_TEST_TRUE_FATAL(wm->GetMetricsRecordRef().HasLabel(METRIC_LABEL_KEY_RUNNER_NAME,
+                                                                  METRIC_LABEL_VALUE_RUNNER_NAME_WATCH_MANAGER));
+
+        // GetHead() is private to WriteMetrics; iterate committed records via the
+        // public DoSnapshot() instead. It returns caller-owned copies of the
+        // undeleted records (deleted ones are already filtered out), so we free
+        // the returned list afterwards to avoid leaking.
+        bool found = false;
+        MetricsRecord* snapshot = WriteMetrics::GetInstance()->DoSnapshot();
+        for (MetricsRecord* record = snapshot; record != nullptr; record = record->GetNext()) {
+            for (auto label = record->GetLabels()->begin(); label != record->GetLabels()->end(); ++label) {
+                if (label->first == METRIC_LABEL_KEY_RUNNER_NAME
+                    && label->second == METRIC_LABEL_VALUE_RUNNER_NAME_WATCH_MANAGER) {
+                    found = true;
+                    break;
+                }
+            }
+            if (found) {
+                break;
+            }
+        }
+        while (snapshot != nullptr) {
+            MetricsRecord* next = snapshot->GetNext();
+            delete snapshot;
+            snapshot = next;
+        }
+        APSARA_TEST_TRUE_FATAL(found);
+    }
+};
+
+APSARA_UNIT_TEST_CASE(WatchManagerUnittest, TestAcquireReleasePassThrough, 0);
+APSARA_UNIT_TEST_CASE(WatchManagerUnittest, TestNoDedupAtWatchManager, 1);
+APSARA_UNIT_TEST_CASE(WatchManagerUnittest, TestSharedEventListenerInstance, 2);
+APSARA_UNIT_TEST_CASE(WatchManagerUnittest, TestMetricsRecordRegistered, 3);
+
+} // namespace logtail
+
+int main(int argc, char** argv) {
+    logtail::Logger::Instance().InitGlobalLoggers();
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}