dym sdk: initialize in module filter after set callbacks (#43660)

Commit Message: dym sdk: initialize in module filter after set callbacks
Additional Description:

By this way, we can access context when creating in module filters.

Risk Level:
Testing:
Docs Changes:
Release Notes:
Platform Specific Features:
[Optional Runtime guard:]
[Optional Fixes #Issue]
[Optional Fixes commit #PR or SHA]
[Optional Deprecated:]
[Optional [API
Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):]

---------

Signed-off-by: wbpcode/wangbaiping <wbphub@gmail.com>

Author: wbpcode

source/extensions/dynamic_modules/BUILD

@@ -28,6 +28,7 @@ envoy_cc_library(
         ":dynamic_modules_lib",
         "//source/common/common:logger_lib",
     ],
+    alwayslink = True,
 )
 
 go_library(

source/extensions/filters/http/dynamic_modules/BUILD

@@ -61,4 +61,5 @@ envoy_cc_library(
         "//source/extensions/dynamic_modules:dynamic_modules_lib",
         "//source/extensions/filters/http/common:pass_through_filter_lib",
     ],
+    alwayslink = True,
 )

source/extensions/filters/http/dynamic_modules/factory.cc

@@ -44,8 +44,13 @@ absl::StatusOr<Http::FilterFactoryCb> DynamicModuleConfigFactory::createFilterFa
     auto filter =
         std::make_shared<Envoy::Extensions::DynamicModules::HttpFilters::DynamicModuleHttpFilter>(
             config, config->stats_scope_->symbolTable());
-    filter->initializeInModuleFilter();
     callbacks.addStreamFilter(filter);
+
+    // The addStreamFilter() will call the setDecoderFilterCallbacks first then
+    // setEncoderFilterCallbacks.
+    // We can initialize the in-module filter after we have both callbacks to ensure the in module
+    // filter can access all the necessary information during creation.
+    filter->initializeInModuleFilter();
   };
 }
 

test/extensions/dynamic_modules/http/BUILD

@@ -61,6 +61,7 @@ envoy_cc_test(
     rbe_pool = "6gig",
     deps = [
         "//source/common/router:string_accessor_lib",
+        "//source/extensions/dynamic_modules:abi_impl",
         "//source/extensions/filters/http/dynamic_modules:abi_impl",
         "//source/extensions/filters/http/dynamic_modules:filter_lib",
         "//test/extensions/dynamic_modules:util",
@@ -78,6 +79,7 @@ envoy_cc_test(
     # http_mocks needs this.
     rbe_pool = "6gig",
     deps = [
+        "//source/extensions/dynamic_modules:abi_impl",
         "//source/extensions/filters/http/dynamic_modules:abi_impl",
         "//test/common/stats:stat_test_utility_lib",
         "//test/mocks/http:http_mocks",
@@ -107,6 +109,8 @@ envoy_cc_test(
     },
     rbe_pool = "6gig",
     deps = [
+        "//source/extensions/dynamic_modules:abi_impl",
+        "//source/extensions/filters/http/dynamic_modules:abi_impl",
         "//source/extensions/filters/http/dynamic_modules:factory_registration",
         "//test/extensions/dynamic_modules:util",
         "//test/integration:http_integration_lib",

test/extensions/dynamic_modules/http/integration_test.cc

@@ -204,6 +204,41 @@ TEST_P(DynamicModulesIntegrationTest, BytesConfig) {
       upstream_request_->headers().get(Http::LowerCaseString("dog"))[0]->value().getStringView());
 }
 
+TEST_P(DynamicModulesIntegrationTest, HeaderCallbacksOnCreation) {
+  initializeFilter("header_callbacks_on_creation", "dog:cat", "",
+                   "type.googleapis.com/google.protobuf.StringValue");
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+
+  Http::TestRequestHeaderMapImpl request_headers{{"foo", "bar"},
+                                                 {":method", "POST"},
+                                                 {":path", "/test/long/url"},
+                                                 {":scheme", "http"},
+                                                 {":authority", "host"}};
+  Http::TestRequestTrailerMapImpl request_trailers{{"foo", "bar"}};
+  Http::TestResponseHeaderMapImpl response_headers{{":status", "200"}, {"foo", "bar"}};
+  Http::TestResponseTrailerMapImpl response_trailers{{"foo", "bar"}};
+
+  auto encoder_decoder = codec_client_->startRequest(request_headers);
+  auto response = std::move(encoder_decoder.second);
+  codec_client_->sendData(encoder_decoder.first, 10, false);
+  codec_client_->sendTrailers(encoder_decoder.first, request_trailers);
+
+  waitForNextUpstreamRequest();
+  // Verify that the headers are added as expected in the filter.
+  EXPECT_EQ(
+      "cat",
+      upstream_request_->headers().get(Http::LowerCaseString("dog"))[0]->value().getStringView());
+  upstream_request_->encodeHeaders(response_headers, false);
+  upstream_request_->encodeData(10, false);
+  upstream_request_->encodeTrailers(response_trailers);
+
+  ASSERT_TRUE(response->waitForEndStream());
+
+  // Verify the proxied request was received upstream, as expected.
+  EXPECT_TRUE(upstream_request_->complete());
+  EXPECT_EQ(10U, upstream_request_->bodyLength());
+}
+
 TEST_P(DynamicModulesIntegrationTest, PerRouteConfig) {
   initializeFilter("per_route_config", "a", "b");
   codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));

test/extensions/dynamic_modules/test_data/cpp/http_integration_test.cc

@@ -323,6 +323,83 @@ class HeaderCallbacksConfigFactory : public HttpFilterConfigFactory {
 
 REGISTER_HTTP_FILTER_CONFIG_FACTORY(HeaderCallbacksConfigFactory, "header_callbacks");
 
+// -----------------------------------------------------------------------------
+// HeaderCallbacksOnCreation
+// -----------------------------------------------------------------------------
+
+class HeaderCallbacksOnCreationFilter : public HttpFilter {
+public:
+  HeadersStatus onRequestHeaders(HeaderMap& headers, bool end_stream) override {
+    return HeadersStatus::Continue;
+  }
+
+  BodyStatus onRequestBody(BodyBuffer& body, bool end_stream) override {
+    return BodyStatus::Continue;
+  }
+
+  TrailersStatus onRequestTrailers(HeaderMap& trailers) override {
+    return TrailersStatus::Continue;
+  }
+
+  HeadersStatus onResponseHeaders(HeaderMap& headers, bool end_stream) override {
+    return HeadersStatus::Continue;
+  }
+
+  BodyStatus onResponseBody(BodyBuffer& body, bool end_stream) override {
+    return BodyStatus::Continue;
+  }
+
+  TrailersStatus onResponseTrailers(HeaderMap& trailers) override {
+    return TrailersStatus::Continue;
+  }
+
+  void onStreamComplete() override {}
+};
+
+class HeaderCallbacksOnCreationFilterFactory : public HttpFilterFactory {
+public:
+  HeaderCallbacksOnCreationFilterFactory(std::map<std::string, std::string> headers_to_add)
+      : headers_to_add_(std::move(headers_to_add)) {}
+
+  std::unique_ptr<HttpFilter> create(HttpFilterHandle& handle) override {
+    for (const auto& [k, v] : headers_to_add_) {
+      handle.requestHeaders().set(k, v);
+    }
+    return std::make_unique<HeaderCallbacksOnCreationFilter>();
+  }
+
+private:
+  std::map<std::string, std::string> headers_to_add_;
+};
+
+class HeaderCallbacksOnCreationConfigFactory : public HttpFilterConfigFactory {
+public:
+  std::unique_ptr<HttpFilterFactory> create(HttpFilterConfigHandle& handle,
+                                            absl::string_view config_view) override {
+    std::map<std::string, std::string> headers_to_add;
+    if (!config_view.empty()) {
+      std::string str(config_view);
+      size_t pos = 0;
+      while ((pos = str.find(',')) != std::string::npos) {
+        std::string part = str.substr(0, pos);
+        size_t sep = part.find(':');
+        if (sep != std::string::npos) {
+          headers_to_add[part.substr(0, sep)] = part.substr(sep + 1);
+        }
+        str.erase(0, pos + 1);
+      }
+      size_t sep = str.find(':');
+      if (sep != std::string::npos) {
+        headers_to_add[str.substr(0, sep)] = str.substr(sep + 1);
+      }
+    }
+    return std::make_unique<HeaderCallbacksOnCreationFilterFactory>(std::move(headers_to_add));
+  }
+};
+
+REGISTER_HTTP_FILTER_CONFIG_FACTORY(HeaderCallbacksOnCreationConfigFactory,
+                                    "header_callbacks_on_creation");
+
 // -----------------------------------------------------------------------------
 // PerRoute
 // -----------------------------------------------------------------------------

test/extensions/dynamic_modules/test_data/go/http_integration_test/http_integration_test.go

@@ -16,20 +16,21 @@ import (
 
 func init() {
 	sdk.RegisterHttpFilterConfigFactories(map[string]shared.HttpFilterConfigFactory{
-		"passthrough":               &PassthroughConfigFactory{},
-		"header_callbacks":          &HeaderCallbacksConfigFactory{},
-		"per_route_config":          &PerRouteConfigFactory{},
-		"body_callbacks":            &BodyCallbacksConfigFactory{},
-		"http_callouts":             &HttpCalloutsConfigFactory{},
-		"send_response":             &SendResponseConfigFactory{},
-		"http_filter_scheduler":     &HttpFilterSchedulerConfigFactory{},
-		"fake_external_cache":       &FakeExternalCacheConfigFactory{},
-		"stats_callbacks":           &StatsCallbacksConfigFactory{},
-		"streaming_terminal_filter": &StreamingTerminalConfigFactory{},
-		"http_stream_basic":         &HttpStreamBasicConfigFactory{},
-		"http_stream_bidirectional": &HttpStreamBidirectionalConfigFactory{},
-		"upstream_reset":            &UpstreamResetConfigFactory{},
-		"http_config_scheduler":     &ConfigSchedulerConfigFactory{},
+		"passthrough":                  &PassthroughConfigFactory{},
+		"header_callbacks":             &HeaderCallbacksConfigFactory{},
+		"header_callbacks_on_creation": &HeaderCallbacksOnCreationConfigFactory{},
+		"per_route_config":             &PerRouteConfigFactory{},
+		"body_callbacks":               &BodyCallbacksConfigFactory{},
+		"http_callouts":                &HttpCalloutsConfigFactory{},
+		"send_response":                &SendResponseConfigFactory{},
+		"http_filter_scheduler":        &HttpFilterSchedulerConfigFactory{},
+		"fake_external_cache":          &FakeExternalCacheConfigFactory{},
+		"stats_callbacks":              &StatsCallbacksConfigFactory{},
+		"streaming_terminal_filter":    &StreamingTerminalConfigFactory{},
+		"http_stream_basic":            &HttpStreamBasicConfigFactory{},
+		"http_stream_bidirectional":    &HttpStreamBidirectionalConfigFactory{},
+		"upstream_reset":               &UpstreamResetConfigFactory{},
+		"http_config_scheduler":        &ConfigSchedulerConfigFactory{},
 	})
 }
 
@@ -178,6 +179,40 @@ type HeaderCallbacksFilter struct {
 	resTrailersCalled bool
 }
 
+type HeaderCallbacksOnCreationConfigFactory struct {
+	shared.EmptyHttpFilterConfigFactory
+}
+
+func (f *HeaderCallbacksOnCreationConfigFactory) Create(handle shared.HttpFilterConfigHandle,
+	config []byte) (shared.HttpFilterFactory, error) {
+	headersToAdd := make(map[string]string)
+	str := string(config)
+	parts := strings.Split(str, ",")
+	for _, part := range parts {
+		kv := strings.Split(part, ":")
+		if len(kv) == 2 {
+			headersToAdd[kv[0]] = kv[1]
+		}
+	}
+	return &HeaderCallbacksOnCreationFilterFactory{headersToAdd: headersToAdd}, nil
+}
+
+type HeaderCallbacksOnCreationFilterFactory struct {
+	shared.EmptyHttpFilterFactory
+	headersToAdd map[string]string
+}
+
+func (f *HeaderCallbacksOnCreationFilterFactory) Create(handle shared.HttpFilterHandle) shared.HttpFilter {
+	for k, v := range f.headersToAdd {
+		handle.RequestHeaders().Set(k, v)
+	}
+	return &HeaderCallbacksOnCreationFilter{}
+}
+
+type HeaderCallbacksOnCreationFilter struct {
+	shared.EmptyHttpFilter
+}
+
 func assert(cond bool, msg string) {
 	if !cond {
 		panic("assertion failed: " + msg)

test/extensions/dynamic_modules/test_data/rust/http_integration_test.rs

@@ -25,6 +25,9 @@ fn new_http_filter_config_fn<EC: EnvoyHttpFilterConfig, EHF: EnvoyHttpFilter>(
     "header_callbacks" => Some(Box::new(HeadersHttpFilterConfig {
       headers_to_add: String::from_utf8(config.to_owned()).unwrap(),
     })),
+    "header_callbacks_on_creation" => Some(Box::new(HeaderCallbacksOnCreationConfig {
+      headers_to_add: String::from_utf8(config.to_owned()).unwrap(),
+    })),
     "per_route_config" => Some(Box::new(PerRouteFilterConfig {
       value: String::from_utf8(config.to_owned()).unwrap(),
     })),
@@ -387,6 +390,26 @@ impl Drop for HeadersHttpFilter {
   }
 }
 
+struct HeaderCallbacksOnCreationConfig {
+  headers_to_add: String,
+}
+
+impl<EHF: EnvoyHttpFilter> HttpFilterConfig<EHF> for HeaderCallbacksOnCreationConfig {
+  fn new_http_filter(&self, envoy: &mut EHF) -> Box<dyn HttpFilter<EHF>> {
+    for header in self.headers_to_add.split(',') {
+      let parts: Vec<&str> = header.split(':').collect();
+      if parts.len() == 2 {
+        envoy.set_request_header(parts[0], parts[1].as_bytes());
+      }
+    }
+    Box::new(HeaderCallbacksOnCreationFilter {})
+  }
+}
+
+struct HeaderCallbacksOnCreationFilter {}
+
+impl<EHF: EnvoyHttpFilter> HttpFilter<EHF> for HeaderCallbacksOnCreationFilter {}
+
 struct PerRouteFilterConfig {
   value: String,
 }