datasource: refactor to use Config::WatchedDirectory directly as it's already exception-free (#41818)

## Description

This PR resolves an old TODO around using `Config::WatchedDirectory`
instead of directly creating a watcher as `Config::WatchedDirectory` is
already exception-free now.

---

**Commit Message:** datasource: refactor to make watched directory
exception-free
**Additional Description:** Use `Config::WatchedDirectory` instead of
directly creating a watcher as `Config::WatchedDirectory` is already
exception-free.
**Risk Level:** Low
**Testing:** Added Tests
**Docs Changes:** N/A
**Release Notes:** N/A

---------

Signed-off-by: Rohit Agrawal <[email protected]>
Co-authored-by: code <[email protected]>

Author: agrawroh

source/common/config/BUILD

@@ -51,6 +51,7 @@ envoy_cc_library(
     deps = [
         ":remote_data_fetcher_lib",
         ":utility_lib",
+        ":watched_directory_lib",
         "//envoy/api:api_interface",
         "//envoy/init:manager_interface",
         "//envoy/thread_local:thread_local_interface",

source/common/config/datasource.cc

@@ -89,7 +89,7 @@ absl::optional<std::string> getPath(const envoy::config::core::v3::DataSource& s
 
 DynamicData::DynamicData(Event::Dispatcher& main_dispatcher,
                          ThreadLocal::TypedSlotPtr<ThreadLocalData> slot,
-                         Filesystem::WatcherPtr watcher)
+                         WatchedDirectoryPtr watcher)
     : dispatcher_(main_dispatcher), slot_(std::move(slot)), watcher_(std::move(watcher)) {}
 
 DynamicData::~DynamicData() {
@@ -140,13 +140,12 @@ absl::StatusOr<DataSourceProviderPtr> DataSourceProvider::create(const ProtoData
   });
 
   const auto& filename = source.filename();
-  auto watcher = main_dispatcher.createFilesystemWatcher();
-  // DynamicData will ensure that the watcher is destroyed before the slot is destroyed.
-  // TODO(wbpcode): use Config::WatchedDirectory instead of directly creating a watcher
-  // if the Config::WatchedDirectory is exception-free in the future.
-  auto watcher_status = watcher->addWatch(
-      absl::StrCat(source.watched_directory().path(), "/"), Filesystem::Watcher::Events::MovedTo,
-      [slot_ptr = slot.get(), &api, filename, allow_empty, max_size](uint32_t) -> absl::Status {
+  auto directory_watcher_or_error =
+      WatchedDirectory::create(source.watched_directory(), main_dispatcher);
+  RETURN_IF_NOT_OK_REF(directory_watcher_or_error.status());
+
+  directory_watcher_or_error.value()->setCallback(
+      [slot_ptr = slot.get(), &api, filename, allow_empty, max_size]() -> absl::Status {
         auto new_data_or_error = readFile(filename, api, allow_empty, max_size);
         if (!new_data_or_error.ok()) {
           // Log an error but don't fail the watch to avoid throwing EnvoyException at runtime.
@@ -163,10 +162,9 @@ absl::StatusOr<DataSourceProviderPtr> DataSourceProvider::create(const ProtoData
             });
         return absl::OkStatus();
       });
-  RETURN_IF_NOT_OK(watcher_status);
 
-  return std::unique_ptr<DataSourceProvider>(
-      new DataSourceProvider(DynamicData(main_dispatcher, std::move(slot), std::move(watcher))));
+  return std::unique_ptr<DataSourceProvider>(new DataSourceProvider(DynamicData(
+      main_dispatcher, std::move(slot), std::move(directory_watcher_or_error.value()))));
 }
 
 } // namespace DataSource

source/common/config/datasource.h

@@ -12,6 +12,7 @@
 #include "source/common/common/empty_string.h"
 #include "source/common/common/enum_to_int.h"
 #include "source/common/config/remote_data_fetcher.h"
+#include "source/common/config/watched_directory.h"
 #include "source/common/init/target_impl.h"
 
 #include "absl/types/optional.h"
@@ -67,15 +68,15 @@ class DynamicData {
 
   DynamicData(DynamicData&&) = default;
   DynamicData(Event::Dispatcher& main_dispatcher, ThreadLocal::TypedSlotPtr<ThreadLocalData> slot,
-              Filesystem::WatcherPtr watcher);
+              WatchedDirectoryPtr watcher);
   ~DynamicData();
 
   const std::string& data() const;
 
 private:
   Event::Dispatcher& dispatcher_;
   ThreadLocal::TypedSlotPtr<ThreadLocalData> slot_;
-  Filesystem::WatcherPtr watcher_;
+  WatchedDirectoryPtr watcher_;
 };
 
 /**

test/common/config/datasource_test.cc

@@ -403,6 +403,48 @@ TEST(DataSourceProviderTest, FileDataSourceAndWithWatchButUpdateError) {
   unlink(TestEnvironment::temporaryPath("envoy_test/watcher_new_link").c_str());
 }
 
+TEST(DataSourceProviderTest, FileDataSourceAndWatchDirectoryCreationFailure) {
+  unlink(TestEnvironment::temporaryPath("envoy_test/watcher_target").c_str());
+  unlink(TestEnvironment::temporaryPath("envoy_test/watcher_link").c_str());
+
+  envoy::config::core::v3::DataSource config;
+  TestEnvironment::createPath(TestEnvironment::temporaryPath("envoy_test"));
+
+  // Use a non-existent directory path that will cause WatchedDirectory::create() to fail.
+  const std::string yaml = fmt::format(R"EOF(
+    filename: "{}"
+    watched_directory:
+      path: "/non/existent/directory/path"
+  )EOF",
+                                       TestEnvironment::temporaryPath("envoy_test/watcher_link"));
+  TestUtility::loadFromYamlAndValidate(yaml, config);
+
+  {
+    std::ofstream file(TestEnvironment::temporaryPath("envoy_test/watcher_target"));
+    file << "Hello, world!";
+    file.close();
+  }
+  TestEnvironment::createSymlink(TestEnvironment::temporaryPath("envoy_test/watcher_target"),
+                                 TestEnvironment::temporaryPath("envoy_test/watcher_link"));
+
+  EXPECT_EQ(envoy::config::core::v3::DataSource::SpecifierCase::kFilename, config.specifier_case());
+
+  Api::ApiPtr api = Api::createApiForTest();
+  Event::DispatcherPtr dispatcher = api->allocateDispatcher("test_thread");
+  NiceMock<ThreadLocal::MockInstance> tls;
+
+  // Creating a provider with an invalid watched directory path should return an error.
+  auto provider_or_error =
+      DataSource::DataSourceProvider::create(config, *dispatcher, tls, *api, false, 0);
+  EXPECT_FALSE(provider_or_error.ok());
+  EXPECT_THAT(provider_or_error.status().message(),
+              testing::HasSubstr("/non/existent/directory/path"));
+
+  // Remove the file.
+  unlink(TestEnvironment::temporaryPath("envoy_test/watcher_target").c_str());
+  unlink(TestEnvironment::temporaryPath("envoy_test/watcher_link").c_str());
+}
+
 } // namespace
 } // namespace Config
 } // namespace Envoy