filesystem: Fix crash when watch callback returns error or throws

Signed-off-by: Rohit Agrawal <rohit.agrawal@databricks.com>

Author: agrawroh

source/common/filesystem/BUILD

@@ -120,12 +120,12 @@ envoy_cc_library(
         "//source/common/common:assert_lib",
         "//source/common/common:linked_object",
         "//source/common/common:minimal_logger_lib",
+        "//source/common/common:thread_lib",
         "//source/common/common:utility_lib",
         "//source/common/network:default_socket_interface_lib",
     ] + select({
         "//bazel:windows_x86_64": [
             "//source/common/api:os_sys_calls_lib",
-            "//source/common/common:thread_lib",
         ],
         "//conditions:default": [],
     }),

source/common/filesystem/inotify/watcher_impl.cc

@@ -10,6 +10,7 @@
 
 #include "source/common/common/assert.h"
 #include "source/common/common/fmt.h"
+#include "source/common/common/thread.h"
 #include "source/common/common/utility.h"
 #include "source/common/filesystem/watcher_impl.h"
 
@@ -52,6 +53,23 @@ absl::Status WatcherImpl::addWatch(absl::string_view path, uint32_t events, OnCh
   return absl::OkStatus();
 }
 
+void WatcherImpl::callAndLogOnError(OnChangedCb& cb, uint32_t events, const std::string& file) {
+  TRY_ASSERT_MAIN_THREAD {
+    const absl::Status status = cb(events);
+    if (!status.ok()) {
+      ENVOY_LOG(warn, "Filesystem watch callback for '{}' returned error: {}", file,
+                status.message());
+    }
+  }
+  END_TRY
+  MULTI_CATCH(
+      const std::exception& e,
+      {
+        ENVOY_LOG(warn, "Filesystem watch callback for '{}' threw exception: {}", file, e.what());
+      },
+      { ENVOY_LOG(warn, "Filesystem watch callback for '{}' threw unknown exception", file); });
+}
+
 absl::Status WatcherImpl::onInotifyEvent() {
   while (true) {
     // The buffer needs to be suitably aligned to store the first inotify_event structure.
@@ -90,10 +108,10 @@ absl::Status WatcherImpl::onInotifyEvent() {
         if (watch.events_ & events) {
           if (watch.file_ == file) {
             ENVOY_LOG(debug, "matched callback: file: {}", file);
-            RETURN_IF_NOT_OK(watch.cb_(events));
+            callAndLogOnError(watch.cb_, events, file);
           } else if (watch.file_.empty()) {
             ENVOY_LOG(debug, "matched callback: directory: {}", file);
-            RETURN_IF_NOT_OK(watch.cb_(events));
+            callAndLogOnError(watch.cb_, events, file);
           }
         }
       }

source/common/filesystem/inotify/watcher_impl.h

@@ -40,6 +40,7 @@ class WatcherImpl : public Watcher, Logger::Loggable<Logger::Id::file> {
   };
 
   absl::Status onInotifyEvent();
+  void callAndLogOnError(OnChangedCb& cb, uint32_t events, const std::string& file);
 
   Filesystem::Instance& file_system_;
   int inotify_fd_;

source/common/filesystem/kqueue/watcher_impl.cc

@@ -8,6 +8,7 @@
 
 #include "source/common/common/assert.h"
 #include "source/common/common/fmt.h"
+#include "source/common/common/thread.h"
 #include "source/common/common/utility.h"
 #include "source/common/filesystem/watcher_impl.h"
 
@@ -116,20 +117,28 @@ absl::Status WatcherImpl::onKqueueEvent() {
 
     absl::StatusOr<PathSplitResult> pathname_or_error =
         file_system_.splitPathFromFilename(file->file_);
-    RETURN_IF_NOT_OK_REF(pathname_or_error.status());
+    if (!pathname_or_error.ok()) {
+      ENVOY_LOG(warn, "Failed to split path '{}': {}", file->file_,
+                pathname_or_error.status().message());
+      continue;
+    }
     PathSplitResult& pathname = pathname_or_error.value();
 
     if (file->watching_dir_) {
       if (event.fflags & NOTE_DELETE) {
-        // directory was deleted
+        // Directory was deleted.
         removeWatch(file);
         return absl::OkStatus();
       }
 
       if (event.fflags & NOTE_WRITE) {
-        // directory was written -- check if the file we're actually watching appeared
+        // Directory was written -- check if the file we're actually watching appeared.
         auto file_or_error = addWatch(file->file_, file->events_, file->callback_, true);
-        RETURN_IF_NOT_OK_REF(file_or_error.status());
+        if (!file_or_error.ok()) {
+          ENVOY_LOG(warn, "Failed to re-add watch for '{}': {}", file->file_,
+                    file_or_error.status().message());
+          continue;
+        }
         FileWatchPtr new_file = file_or_error.value();
         if (new_file != nullptr) {
           removeWatch(file);
@@ -150,7 +159,11 @@ absl::Status WatcherImpl::onKqueueEvent() {
         removeWatch(file);
 
         auto file_or_error = addWatch(file->file_, file->events_, file->callback_, true);
-        RETURN_IF_NOT_OK_REF(file_or_error.status());
+        if (!file_or_error.ok()) {
+          ENVOY_LOG(warn, "Failed to re-add watch for '{}': {}", file->file_,
+                    file_or_error.status().message());
+          continue;
+        }
         FileWatchPtr new_file = file_or_error.value();
         if (new_file == nullptr) {
           return absl::OkStatus();
@@ -173,11 +186,29 @@ absl::Status WatcherImpl::onKqueueEvent() {
 
     if (events & file->events_) {
       ENVOY_LOG(debug, "matched callback: file: {}", file->file_);
-      RETURN_IF_NOT_OK(file->callback_(events));
+      callAndLogOnError(file->callback_, events, file->file_);
     }
   }
   return absl::OkStatus();
 }
 
+void WatcherImpl::callAndLogOnError(Watcher::OnChangedCb& cb, uint32_t events,
+                                    const std::string& file) {
+  TRY_ASSERT_MAIN_THREAD {
+    const absl::Status status = cb(events);
+    if (!status.ok()) {
+      ENVOY_LOG(warn, "Filesystem watch callback for '{}' returned error: {}", file,
+                status.message());
+    }
+  }
+  END_TRY
+  MULTI_CATCH(
+      const std::exception& e,
+      {
+        ENVOY_LOG(warn, "Filesystem watch callback for '{}' threw exception: {}", file, e.what());
+      },
+      { ENVOY_LOG(warn, "Filesystem watch callback for '{}' threw unknown exception", file); });
+}
+
 } // namespace Filesystem
 } // namespace Envoy

source/common/filesystem/kqueue/watcher_impl.h

@@ -46,6 +46,7 @@ class WatcherImpl : public Watcher, Logger::Loggable<Logger::Id::file> {
   absl::StatusOr<FileWatchPtr> addWatch(absl::string_view path, uint32_t events,
                                         Watcher::OnChangedCb cb, bool pathMustExist);
   void removeWatch(FileWatchPtr& watch);
+  void callAndLogOnError(OnChangedCb& cb, uint32_t events, const std::string& file);
 
   Filesystem::Instance& file_system_;
   int queue_;