[Wasm] Add cluster metadata fallback and upstream host metadata (envo…

…yproxy#13477) (#275)

Signed-off-by: Pengyuan Bian <bianpengyuan@google.com>

source/extensions/common/wasm/context.cc

@@ -403,7 +403,8 @@ WasmResult serializeValue(Filters::Common::Expr::CelValue value, std::string* re
   _f(METADATA) _f(REQUEST) _f(RESPONSE) _f(CONNECTION) _f(UPSTREAM) _f(NODE) _f(SOURCE)            \
       _f(DESTINATION) _f(LISTENER_DIRECTION) _f(LISTENER_METADATA) _f(CLUSTER_NAME)                \
           _f(CLUSTER_METADATA) _f(ROUTE_NAME) _f(ROUTE_METADATA) _f(PLUGIN_NAME)                   \
-              _f(PLUGIN_ROOT_ID) _f(PLUGIN_VM_ID) _f(CONNECTION_ID) _f(FILTER_STATE)
+              _f(UPSTREAM_HOST_METADATA) _f(PLUGIN_ROOT_ID) _f(PLUGIN_VM_ID) _f(CONNECTION_ID)     \
+                  _f(FILTER_STATE)
 
 static inline std::string downCase(std::string s) {
   std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return std::tolower(c); });
@@ -524,6 +525,14 @@ Context::findValue(absl::string_view name, Protobuf::Arena* arena, bool last) co
   case PropertyToken::CLUSTER_METADATA:
     if (info && info->upstreamHost()) {
       return CelValue::CreateMessage(&info->upstreamHost()->cluster().metadata(), arena);
+    } else if (info && info->upstreamClusterInfo().has_value() &&
+               info->upstreamClusterInfo().value()) {
+      return CelValue::CreateMessage(&info->upstreamClusterInfo().value()->metadata(), arena);
+    }
+    break;
+  case PropertyToken::UPSTREAM_HOST_METADATA:
+    if (info && info->upstreamHost() && info->upstreamHost()->metadata()) {
+      return CelValue::CreateMessage(info->upstreamHost()->metadata().get(), arena);
     }
     break;
   case PropertyToken::ROUTE_NAME:

test/extensions/filters/http/wasm/test_data/test_cpp.cc

@@ -328,6 +328,11 @@ void TestContext::onLog() {
     if (response_trailer && response_trailer->view() != "") {
       logWarn("response bogus-trailer found");
     }
+  } else if (test == "cluster_metadata") {
+      std::string cluster_metadata;
+      if (getValue({"cluster_metadata", "filter_metadata", "namespace", "key"}, &cluster_metadata)) {
+        logWarn("cluster metadata: " + cluster_metadata);
+      }
   } else if (test == "property") {
     setFilterState("wasm_state", "wasm_value");
     auto path = getRequestHeader(":path");
@@ -343,6 +348,10 @@ void TestContext::onLog() {
       if (getValue({"response", "code"}, &responseCode)) {
         logWarn("response.code: " + std::to_string(responseCode));
       }
+      std::string upstream_host_metadata;
+      if (getValue({"upstream_host_metadata", "filter_metadata", "namespace", "key"}, &upstream_host_metadata)) {
+        logWarn("upstream host metadata: " + upstream_host_metadata);
+      }
       logWarn("state: " + getProperty({"wasm_state"}).value()->toString());
     } else {
       logWarn("onLog " + std::to_string(id()) + " " + std::string(path->view()));
@@ -541,7 +550,6 @@ void TestContext::onLog() {
           {{"source", "address"}, "127.0.0.1:0"},
           {{"destination", "address"}, "127.0.0.2:0"},
           {{"upstream", "address"}, "10.0.0.1:443"},
-          {{"cluster_metadata"}, ""},
           {{"route_metadata"}, ""},
       };
       for (const auto& property : properties) {

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

@@ -1294,6 +1294,8 @@ TEST_P(WasmHttpFilterTest, Property) {
               log_(spdlog::level::warn, Eq(absl::string_view("metadata: wasm_request_get_value"))));
   EXPECT_CALL(filter(), log_(spdlog::level::warn, Eq(absl::string_view("response.code: 403"))));
   EXPECT_CALL(filter(), log_(spdlog::level::warn, Eq(absl::string_view("state: wasm_value"))));
+  EXPECT_CALL(filter(),
+              log_(spdlog::level::warn, Eq(absl::string_view("upstream host metadata: endpoint"))));
 
   root_context_->onTick(0);
   Http::TestRequestHeaderMapImpl request_headers{{":path", "/test_context"}};
@@ -1306,6 +1308,54 @@ TEST_P(WasmHttpFilterTest, Property) {
   EXPECT_CALL(encoder_callbacks_, connection()).WillRepeatedly(Return(&connection));
   NiceMock<Router::MockRouteEntry> route_entry;
   EXPECT_CALL(request_stream_info_, routeEntry()).WillRepeatedly(Return(&route_entry));
+  std::shared_ptr<NiceMock<Envoy::Upstream::MockHostDescription>> host_description(
+      new NiceMock<Envoy::Upstream::MockHostDescription>());
+  auto metadata = std::make_shared<envoy::config::core::v3::Metadata>(
+      TestUtility::parseYaml<envoy::config::core::v3::Metadata>(
+          R"EOF(
+        filter_metadata:
+          namespace:
+            key: endpoint
+      )EOF"));
+  EXPECT_CALL(*host_description, metadata()).WillRepeatedly(Return(metadata));
+  EXPECT_CALL(request_stream_info_, upstreamHost()).WillRepeatedly(Return(host_description));
+  filter().log(&request_headers, nullptr, nullptr, log_stream_info);
+}
+
+TEST_P(WasmHttpFilterTest, ClusterMetadata) {
+  if (std::get<1>(GetParam()) == "rust") {
+    // TODO(PiotrSikora): test not yet implemented using Rust SDK.
+    return;
+  }
+  setupTest("", "cluster_metadata");
+  setupFilter();
+  EXPECT_CALL(filter(),
+              log_(spdlog::level::warn, Eq(absl::string_view("cluster metadata: cluster"))));
+  auto cluster = std::make_shared<NiceMock<Upstream::MockClusterInfo>>();
+  auto cluster_metadata = std::make_shared<envoy::config::core::v3::Metadata>(
+      TestUtility::parseYaml<envoy::config::core::v3::Metadata>(
+          R"EOF(
+      filter_metadata:
+        namespace:
+          key: cluster
+    )EOF"));
+
+  std::shared_ptr<NiceMock<Envoy::Upstream::MockHostDescription>> host_description(
+      new NiceMock<Envoy::Upstream::MockHostDescription>());
+  StreamInfo::MockStreamInfo log_stream_info;
+  Http::TestRequestHeaderMapImpl request_headers{{}};
+
+  EXPECT_CALL(encoder_callbacks_, streamInfo()).WillRepeatedly(ReturnRef(request_stream_info_));
+  EXPECT_CALL(*cluster, metadata()).WillRepeatedly(ReturnRef(*cluster_metadata));
+  EXPECT_CALL(*host_description, cluster()).WillRepeatedly(ReturnRef(*cluster));
+  EXPECT_CALL(request_stream_info_, upstreamHost()).WillRepeatedly(Return(host_description));
+  filter().log(&request_headers, nullptr, nullptr, log_stream_info);
+
+  // If upstream host is empty, fallback to upstream cluster info for cluster metadata.
+  EXPECT_CALL(request_stream_info_, upstreamHost()).WillRepeatedly(Return(nullptr));
+  EXPECT_CALL(request_stream_info_, upstreamClusterInfo()).WillRepeatedly(Return(cluster));
+  EXPECT_CALL(filter(),
+              log_(spdlog::level::warn, Eq(absl::string_view("cluster metadata: cluster"))));
   filter().log(&request_headers, nullptr, nullptr, log_stream_info);
 }