Allow prepending access log handler for WASM filter

changelogs/current.yaml

@@ -39,6 +39,14 @@ behavior_changes:
   change: |
     Added new tag extraction so that scoped rds stats have their :ref:'scope_route_config_name
     <envoy_v3_api_msg_config/route/v3/scoped_route>' and stat prefix extracted.
+- area: wasm
+  change: |
+    Added prependAccessLogHandler in HTTP filter to be used by WASM in order to fix
+    `issue 30859 <https://github.com/envoyproxy/envoy/issues/30859>`_, which allows
+    reversing adding the order of access loggers in order to provide correct dynamic metadata flow
+    without inheriting onStreamComplete(). This solution is exclusively used by WASM, and may be
+    changed to an updated WASM access logger if additional features deems it necessary. This behavior can
+    be reverted by setting the runtime guard ``envoy.reloadable_features.prepend_access_log_handler`` to false.
 
 minor_behavior_changes:
 # *Changes that may cause incompatibilities for some users, but should not for most*

envoy/http/filter.h

@@ -1247,6 +1247,13 @@ class FilterChainFactoryCallbacks {
    */
   virtual void addAccessLogHandler(AccessLog::InstanceSharedPtr handler) PURE;
 
+  /**
+   * Prepend an access log handler that is called when the stream is destroyed.
+   * @param handler supplies the handler to add. Currently used specifically for
+   * WASM filter.
+   */
+  virtual void prependAccessLogHandler(AccessLog::InstanceSharedPtr handler) PURE;
+
   /**
    * Allows filters to access the thread local dispatcher.
    * @param return the worker thread's dispatcher.

source/common/http/filter_manager.h

@@ -676,6 +676,9 @@ class FilterManager : public ScopeTrackedObject,
   void addAccessLogHandler(AccessLog::InstanceSharedPtr handler) {
     access_log_handlers_.push_back(std::move(handler));
   }
+  void prependAccessLogHandler(AccessLog::InstanceSharedPtr handler) {
+    access_log_handlers_.push_front(std::move(handler));
+  }
   void addStreamDecoderFilter(ActiveStreamDecoderFilterPtr filter) {
     // Note: configured decoder filters are appended to decoder_filters_.
     // This means that if filters are configured in the following order (assume all three filters
@@ -975,6 +978,9 @@ class FilterManager : public ScopeTrackedObject,
     void addAccessLogHandler(AccessLog::InstanceSharedPtr handler) override {
       manager_.addAccessLogHandler(std::move(handler));
     }
+    void prependAccessLogHandler(AccessLog::InstanceSharedPtr handler) override {
+      manager_.prependAccessLogHandler(std::move(handler));
+    }
 
     Event::Dispatcher& dispatcher() override { return manager_.dispatcher_; }
 

source/common/runtime/runtime_features.cc

@@ -72,6 +72,7 @@ RUNTIME_GUARD(envoy_reloadable_features_no_extension_lookup_by_name);
 RUNTIME_GUARD(envoy_reloadable_features_no_timer_based_rate_limit_token_bucket);
 RUNTIME_GUARD(envoy_reloadable_features_original_dst_rely_on_idle_timeout);
 RUNTIME_GUARD(envoy_reloadable_features_prefer_ipv6_dns_on_macos);
+RUNTIME_GUARD(envoy_reloadable_features_prepend_access_log_handler);
 RUNTIME_GUARD(envoy_reloadable_features_proxy_104);
 RUNTIME_GUARD(envoy_reloadable_features_proxy_status_mapping_more_core_response_flags);
 RUNTIME_GUARD(envoy_reloadable_features_quic_connect_client_udp_sockets);

source/extensions/filters/http/composite/factory_wrapper.cc

@@ -42,6 +42,9 @@ void FactoryCallbacksWrapper::addStreamFilter(Http::StreamFilterSharedPtr filter
 void FactoryCallbacksWrapper::addAccessLogHandler(AccessLog::InstanceSharedPtr access_log) {
   access_loggers_.push_back(std::move(access_log));
 }
+void FactoryCallbacksWrapper::prependAccessLogHandler(AccessLog::InstanceSharedPtr access_log) {
+  access_loggers_.push_front(std::move(access_log));
+}
 } // namespace Composite
 } // namespace HttpFilters
 } // namespace Extensions

source/extensions/filters/http/composite/factory_wrapper.h

@@ -21,6 +21,7 @@ struct FactoryCallbacksWrapper : public Http::FilterChainFactoryCallbacks {
   void addStreamEncoderFilter(Http::StreamEncoderFilterSharedPtr filter) override;
   void addStreamFilter(Http::StreamFilterSharedPtr filter) override;
   void addAccessLogHandler(AccessLog::InstanceSharedPtr) override;
+  void prependAccessLogHandler(AccessLog::InstanceSharedPtr) override;
   Event::Dispatcher& dispatcher() override { return dispatcher_; }
 
   Filter& filter_;
@@ -30,7 +31,7 @@ struct FactoryCallbacksWrapper : public Http::FilterChainFactoryCallbacks {
       absl::variant<Http::StreamDecoderFilterSharedPtr, Http::StreamEncoderFilterSharedPtr,
                     Http::StreamFilterSharedPtr>;
   absl::optional<FilterAlternative> filter_to_inject_;
-  std::vector<AccessLog::InstanceSharedPtr> access_loggers_;
+  std::list<AccessLog::InstanceSharedPtr> access_loggers_;
   std::vector<absl::Status> errors_;
 };
 } // namespace Composite

source/extensions/filters/http/match_delegate/config.cc

@@ -92,6 +92,10 @@ struct DelegatingFactoryCallbacks : public Envoy::Http::FilterChainFactoryCallba
     delegated_callbacks_.addAccessLogHandler(std::move(handler));
   }
 
+  void prependAccessLogHandler(AccessLog::InstanceSharedPtr handler) override {
+    delegated_callbacks_.prependAccessLogHandler(std::move(handler));
+  }
+
   Envoy::Http::FilterChainFactoryCallbacks& delegated_callbacks_;
   Matcher::MatchTreeSharedPtr<Envoy::Http::HttpMatchingData> match_tree_;
 };

source/extensions/filters/http/wasm/config.h

@@ -60,7 +60,11 @@ class WasmFilterConfig
         return;
       }
       callbacks.addStreamFilter(filter);
-      callbacks.addAccessLogHandler(filter);
+      if (Runtime::runtimeFeatureEnabled("envoy.reloadable_features.prepend_access_log_handler")) {
+        callbacks.prependAccessLogHandler(filter);
+      } else {
+        callbacks.addAccessLogHandler(filter);
+      }
     };
   }
 };

test/extensions/filters/http/wasm/config_test.cc

@@ -172,7 +172,7 @@ TEST_P(WasmFilterConfigTest, JsonLoadFromFileWasm) {
   EXPECT_EQ(getContextInitManagerState(), Init::Manager::State::Initialized);
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
 }
 
@@ -208,7 +208,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromFileWasm) {
         .WillOnce([&context](Http::StreamFilterSharedPtr filter) {
           context = std::static_pointer_cast<Envoy::Extensions::Common::Wasm::Context>(filter);
         });
-    EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+    EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
     cb(filter_callback);
   }
   // Check if the context still holds a valid Wasm even after the factory is destroyed.
@@ -242,7 +242,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromFileWasmFailOpenOk) {
   EXPECT_EQ(getContextInitManagerState(), Init::Manager::State::Initialized);
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
 }
 
@@ -292,7 +292,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromFileWasmInvalidConfig) {
   EXPECT_EQ(getContextInitManagerState(), Init::Manager::State::Initialized);
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
 }
 
@@ -318,7 +318,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadInlineWasm) {
   EXPECT_EQ(getContextInitManagerState(), Init::Manager::State::Initialized);
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
 }
 
@@ -385,7 +385,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteWasm) {
   EXPECT_EQ(getContextInitManagerState(), Init::Manager::State::Initialized);
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
 }
 
@@ -449,7 +449,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteWasmFailOnUncachedThenSucceed) {
 
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
 
   cb(filter_callback);
   dispatcher_.clearDeferredDeleteList();
@@ -572,7 +572,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteWasmFailCachedThenSucceed) {
 
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
 
   cb(filter_callback);
 
@@ -639,7 +639,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteWasmFailCachedThenSucceed) {
 
   Http::MockFilterChainFactoryCallbacks filter_callback2;
   EXPECT_CALL(filter_callback2, addStreamFilter(_));
-  EXPECT_CALL(filter_callback2, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback2, prependAccessLogHandler(_));
 
   cb2(filter_callback2);
 
@@ -847,7 +847,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteMultipleRetries) {
   EXPECT_EQ(getContextInitManagerState(), Init::Manager::State::Initialized);
   Http::MockFilterChainFactoryCallbacks filter_callback;
   EXPECT_CALL(filter_callback, addStreamFilter(_));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
 }
 
@@ -900,7 +900,7 @@ TEST_P(WasmFilterConfigTest, YamlLoadFromRemoteSuccessBadcode) {
       .WillOnce(Invoke([&context](Http::StreamFilterSharedPtr filter) {
         context = std::static_pointer_cast<Extensions::Common::Wasm::Context>(filter);
       }));
-  EXPECT_CALL(filter_callback, addAccessLogHandler(_));
+  EXPECT_CALL(filter_callback, prependAccessLogHandler(_));
   cb(filter_callback);
   EXPECT_EQ(context->wasm(), nullptr);
   EXPECT_TRUE(context->isFailed());

test/mocks/http/mocks.h

@@ -182,6 +182,7 @@ class MockFilterChainFactoryCallbacks : public Http::FilterChainFactoryCallbacks
   MOCK_METHOD(void, addStreamEncoderFilter, (Http::StreamEncoderFilterSharedPtr filter));
   MOCK_METHOD(void, addStreamFilter, (Http::StreamFilterSharedPtr filter));
   MOCK_METHOD(void, addAccessLogHandler, (AccessLog::InstanceSharedPtr handler));
+  MOCK_METHOD(void, prependAccessLogHandler, (AccessLog::InstanceSharedPtr handler));
   MOCK_METHOD(Event::Dispatcher&, dispatcher, ());
 };