data source: new data source provider to support file watching (#33390)

This PR provide a common data source provider to support file watching.

For the users who don't need the file watching or don't use the file data source, if the provider is used, then only need to pay 8 additional bytes and one additional if check (holds_alternative) compare to using the directly DataSource::read().

For the users who want to use the file watching, additional file watcher and TLS slot (ThreadLocalStorage) are necessary. This is much expensive but reasonable.

Risk Level: low.
Testing: unit.
Docs Changes: n/a.
Release Notes: n/a.
Platform Specific Features: n/a.

Signed-off-by: wbpcode <[email protected]>
Signed-off-by: code <[email protected]>

api/envoy/config/core/v3/base.proto

@@ -411,6 +411,7 @@ message WatchedDirectory {
 }
 
 // Data source consisting of a file, an inline value, or an environment variable.
+// [#next-free-field: 6]
 message DataSource {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.core.DataSource";
 
@@ -429,6 +430,22 @@ message DataSource {
     // Environment variable data source.
     string environment_variable = 4 [(validate.rules).string = {min_len: 1}];
   }
+
+  // Watched directory that is watched for file changes. If this is set explicitly, the file
+  // specified in the ``filename`` field will be reloaded when relevant file move events occur.
+  //
+  // .. note::
+  //   This field only makes sense when the ``filename`` field is set.
+  //
+  // .. note::
+  //   Envoy only updates when the file is replaced by a file move, and not when the file is
+  //   edited in place.
+  //
+  // .. note::
+  //   Not all use cases of ``DataSource`` support watching directories. It depends on the
+  //   specific usage of the ``DataSource``. See the documentation of the parent message for
+  //   details.
+  WatchedDirectory watched_directory = 5;
 }
 
 // The message specifies the retry policy of remote data source when fetching fails.

source/common/config/BUILD

@@ -53,6 +53,7 @@ envoy_cc_library(
         ":utility_lib",
         "//envoy/api:api_interface",
         "//envoy/init:manager_interface",
+        "//envoy/thread_local:thread_local_interface",
         "//envoy/upstream:cluster_manager_interface",
         "//source/common/common:backoff_lib",
         "//source/common/common:empty_string",

source/common/config/datasource.cc

@@ -13,27 +13,9 @@ namespace DataSource {
 absl::StatusOr<std::string> read(const envoy::config::core::v3::DataSource& source,
                                  bool allow_empty, Api::Api& api, uint64_t max_size) {
   std::string data;
-  absl::StatusOr<std::string> file_or_error;
   switch (source.specifier_case()) {
   case envoy::config::core::v3::DataSource::SpecifierCase::kFilename:
-    if (max_size > 0) {
-      if (!api.fileSystem().fileExists(source.filename())) {
-        return absl::InvalidArgumentError(fmt::format("file {} does not exist", source.filename()));
-      }
-      const ssize_t size = api.fileSystem().fileSize(source.filename());
-      if (size < 0) {
-        return absl::InvalidArgumentError(
-            absl::StrCat("cannot determine size of file ", source.filename()));
-      }
-      if (static_cast<uint64_t>(size) > max_size) {
-        return absl::InvalidArgumentError(fmt::format("file {} size is {} bytes; maximum is {}",
-                                                      source.filename(), size, max_size));
-      }
-    }
-    file_or_error = api.fileSystem().fileReadToEnd(source.filename());
-    RETURN_IF_STATUS_NOT_OK(file_or_error);
-    data = file_or_error.value();
-    break;
+    return readFile(source.filename(), api, allow_empty, max_size);
   case envoy::config::core::v3::DataSource::SpecifierCase::kInlineBytes:
     data = source.inline_bytes();
     break;
@@ -61,12 +43,111 @@ absl::StatusOr<std::string> read(const envoy::config::core::v3::DataSource& sour
   return data;
 }
 
+absl::StatusOr<std::string> readFile(const std::string& path, Api::Api& api, bool allow_empty,
+                                     uint64_t max_size) {
+  auto& file_system = api.fileSystem();
+
+  if (max_size > 0) {
+    if (!file_system.fileExists(path)) {
+      return absl::InvalidArgumentError(fmt::format("file {} does not exist", path));
+    }
+    const ssize_t size = file_system.fileSize(path);
+    if (size < 0) {
+      return absl::InvalidArgumentError(absl::StrCat("cannot determine size of file ", path));
+    }
+    if (static_cast<uint64_t>(size) > max_size) {
+      return absl::InvalidArgumentError(
+          fmt::format("file {} size is {} bytes; maximum is {}", path, size, max_size));
+    }
+  }
+
+  auto file_content_or_error = file_system.fileReadToEnd(path);
+  RETURN_IF_STATUS_NOT_OK(file_content_or_error);
+
+  if (!allow_empty && file_content_or_error.value().empty()) {
+    return absl::InvalidArgumentError(fmt::format("file {} is empty", path));
+  }
+
+  return file_content_or_error.value();
+}
+
 absl::optional<std::string> getPath(const envoy::config::core::v3::DataSource& source) {
   return source.specifier_case() == envoy::config::core::v3::DataSource::SpecifierCase::kFilename
              ? absl::make_optional(source.filename())
              : absl::nullopt;
 }
 
+DynamicData::DynamicData(Event::Dispatcher& main_dispatcher,
+                         ThreadLocal::TypedSlotPtr<ThreadLocalData> slot,
+                         Filesystem::WatcherPtr watcher)
+    : dispatcher_(main_dispatcher), slot_(std::move(slot)), watcher_(std::move(watcher)) {}
+
+DynamicData::~DynamicData() {
+  if (!dispatcher_.isThreadSafe()) {
+    dispatcher_.post([to_delete = std::move(slot_)] {});
+  }
+}
+
+absl::string_view DynamicData::data() const {
+  const auto thread_local_data = slot_->get();
+  return thread_local_data.has_value() ? *thread_local_data->data_ : EMPTY_STRING;
+}
+
+absl::string_view DataSourceProvider::data() const {
+  if (absl::holds_alternative<std::string>(data_)) {
+    return absl::get<std::string>(data_);
+  }
+  return absl::get<DynamicData>(data_).data();
+}
+
+absl::StatusOr<DataSourceProvider> DataSourceProvider::create(const ProtoDataSource& source,
+                                                              Event::Dispatcher& main_dispatcher,
+                                                              ThreadLocal::SlotAllocator& tls,
+                                                              Api::Api& api, bool allow_empty,
+                                                              uint64_t max_size) {
+  auto initial_data_or_error = read(source, allow_empty, api, max_size);
+  RETURN_IF_STATUS_NOT_OK(initial_data_or_error);
+
+  if (!source.has_watched_directory() ||
+      source.specifier_case() != envoy::config::core::v3::DataSource::kFilename) {
+    return DataSourceProvider(std::move(initial_data_or_error).value());
+  }
+
+  auto slot = ThreadLocal::TypedSlot<DynamicData::ThreadLocalData>::makeUnique(tls);
+  slot->set([initial_data = std::make_shared<std::string>(
+                 std::move(initial_data_or_error.value()))](Event::Dispatcher&) {
+    return std::make_shared<DynamicData::ThreadLocalData>(initial_data);
+  });
+
+  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 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.
+          ENVOY_LOG_TO_LOGGER(Logger::Registry::getLog(Logger::Id::config), error,
+                              "Failed to read file: {}", new_data_or_error.status().message());
+          return absl::OkStatus();
+        }
+        slot_ptr->runOnAllThreads(
+            [new_data = std::make_shared<std::string>(std::move(new_data_or_error.value()))](
+                OptRef<DynamicData::ThreadLocalData> obj) {
+              if (obj.has_value()) {
+                obj->data_ = new_data;
+              }
+            });
+        return absl::OkStatus();
+      });
+  RETURN_IF_NOT_OK(watcher_status);
+
+  return DataSourceProvider(DynamicData(main_dispatcher, std::move(slot), std::move(watcher)));
+}
+
 } // namespace DataSource
 } // namespace Config
 } // namespace Envoy

source/common/config/datasource.h

@@ -5,6 +5,7 @@
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/event/deferred_deletable.h"
 #include "envoy/init/manager.h"
+#include "envoy/thread_local/thread_local.h"
 #include "envoy/upstream/cluster_manager.h"
 
 #include "source/common/common/backoff_strategy.h"
@@ -19,25 +20,98 @@ namespace Envoy {
 namespace Config {
 namespace DataSource {
 
+using ProtoDataSource = envoy::config::core::v3::DataSource;
+using ProtoWatchedDirectory = envoy::config::core::v3::WatchedDirectory;
+
 /**
  * Read contents of the DataSource.
  * @param source data source.
  * @param allow_empty return an empty string if no DataSource case is specified.
- * @param api reference to the Api object
+ * @param api reference to the Api.
  * @param max_size max size limit of file to read, default 0 means no limit, and if the file data
- * would exceed the limit, it will throw a EnvoyException.
+ * would exceed the limit, it will return an error status.
  * @return std::string with DataSource contents. or an error status if no DataSource case is
  * specified and !allow_empty.
  */
 absl::StatusOr<std::string> read(const envoy::config::core::v3::DataSource& source,
                                  bool allow_empty, Api::Api& api, uint64_t max_size = 0);
 
+/**
+ * Read contents of the file.
+ * @param path file path.
+ * @param api reference to the Api.
+ * @param allow_empty return an empty string if the file is empty.
+ * @param max_size max size limit of file to read, default 0 means no limit, and if the file data
+ * would exceed the limit, it will return an error status.
+ * @return std::string with file contents. or an error status if the file does not exist or
+ * cannot be read.
+ */
+absl::StatusOr<std::string> readFile(const std::string& path, Api::Api& api, bool allow_empty,
+                                     uint64_t max_size = 0);
+
 /**
  * @param source data source.
  * @return absl::optional<std::string> path to DataSource if a filename, otherwise absl::nullopt.
  */
 absl::optional<std::string> getPath(const envoy::config::core::v3::DataSource& source);
 
+class DynamicData {
+public:
+  struct ThreadLocalData : public ThreadLocal::ThreadLocalObject {
+    ThreadLocalData(std::shared_ptr<std::string> data) : data_(std::move(data)) {}
+    std::shared_ptr<std::string> data_;
+  };
+
+  DynamicData(DynamicData&&) = default;
+  DynamicData(Event::Dispatcher& main_dispatcher, ThreadLocal::TypedSlotPtr<ThreadLocalData> slot,
+              Filesystem::WatcherPtr watcher);
+  ~DynamicData();
+
+  absl::string_view data() const;
+
+private:
+  Event::Dispatcher& dispatcher_;
+  ThreadLocal::TypedSlotPtr<ThreadLocalData> slot_;
+  Filesystem::WatcherPtr watcher_;
+};
+
+/**
+ * DataSourceProvider provides a way to get the DataSource contents and watch the possible
+ * content changes. The watch only works for filename-based DataSource and watched directory
+ * is provided explicitly.
+ *
+ * NOTE: This should only be used when the envoy.config.core.v3.DataSource is necessary and
+ * file watch is required.
+ */
+class DataSourceProvider {
+public:
+  /**
+   * Create a DataSourceProvider from a DataSource.
+   * @param source data source.
+   * @param main_dispatcher reference to the main dispatcher.
+   * @param tls reference to the thread local slot allocator.
+   * @param api reference to the Api.
+   * @param allow_empty return an empty string if no DataSource case is specified.
+   * @param max_size max size limit of file to read, default 0 means no limit.
+   * @return absl::StatusOr<DataSourceProvider> with DataSource contents. or an error
+   * status if any error occurs.
+   * NOTE: If file watch is enabled and the new file content does not meet the
+   * requirements (allow_empty, max_size), the provider will keep the old content.
+   */
+  static absl::StatusOr<DataSourceProvider> create(const ProtoDataSource& source,
+                                                   Event::Dispatcher& main_dispatcher,
+                                                   ThreadLocal::SlotAllocator& tls, Api::Api& api,
+                                                   bool allow_empty, uint64_t max_size = 0);
+
+  absl::string_view data() const;
+
+private:
+  DataSourceProvider(std::string&& data) : data_(std::move(data)) {}
+  DataSourceProvider(DynamicData&& data) : data_(std::move(data)) {}
+
+  absl::variant<std::string, DynamicData> data_;
+};
+
 } // namespace DataSource
 } // namespace Config
 } // namespace Envoy

test/common/config/BUILD

@@ -199,6 +199,7 @@ envoy_cc_test(
         "//test/mocks/event:event_mocks",
         "//test/mocks/init:init_mocks",
         "//test/mocks/runtime:runtime_mocks",
+        "//test/mocks/thread_local:thread_local_mocks",
         "//test/mocks/upstream:cluster_manager_mocks",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",