extensions/thrift_proxy: Add header matching to thrift router (envoyproxy#4239)

This change adds header matching to the thrift router We do this by pulling in the route proto definition into the thrift route proto and making use of the Http::HeaderUtility class to do the matching for us. As such, we support the same type of header matching that exists for the http router.

Risk Level: LOW
Testing: unit and integrations tests, new and old, pass.
Doc changes: api docs updated
Release notes: n/a

Signed-off-by: Brian Ramos <[email protected]>

Author: brirams

api/envoy/config/filter/network/thrift_proxy/v2alpha1/BUILD

@@ -8,4 +8,7 @@ api_proto_library_internal(
         "route.proto",
         "thrift_proxy.proto",
     ],
+    deps = [
+        "//envoy/api/v2/route",
+    ],
 )

api/envoy/config/filter/network/thrift_proxy/v2alpha1/route.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package envoy.config.filter.network.thrift_proxy.v2alpha1;
 option go_package = "v2";
 
+import "envoy/api/v2/route/route.proto";
 import "validate/validate.proto";
 import "gogoproto/gogo.proto";
 
@@ -28,7 +29,7 @@ message Route {
   RouteAction route = 2 [(validate.rules).message.required = true, (gogoproto.nullable) = false];
 }
 
-// [#comment:next free field: 4]
+// [#comment:next free field: 5]
 message RouteMatch {
   oneof match_specifier {
     option (validate.required) = true;
@@ -43,9 +44,26 @@ message RouteMatch {
     string service_name = 2;
   }
 
-  // Inverts whatever matching is done in match_specifier. Cannot be combined with wildcard matching
-  // as that would result in routes never being matched.
+  // Inverts whatever matching is done in the :ref:`method_name
+  // <envoy_api_field_config.filter.network.thrift_proxy.v2alpha1.RouteMatch.method_name>` or
+  // :ref:`service_name
+  // <envoy_api_field_config.filter.network.thrift_proxy.v2alpha1.RouteMatch.service_name>` fields.
+  // Cannot be combined with wildcard matching as that would result in routes never being matched.
+  //
+  // .. note::
+  //
+  //   This does not invert matching done as part of the :ref:`headers field
+  //   <envoy_api_field_config.filter.network.thrift_proxy.v2alpha1.RouteMatch.headers>` field. To
+  //   invert header matching, see :ref:`invert_match
+  //   <envoy_api_field_route.HeaderMatcher.invert_match>`.
   bool invert = 3;
+
+  // Specifies a set of headers that the route should match on. The router will check the request’s
+  // headers against all the specified headers in the route config. A match will happen if all the
+  // headers in the route are present in the request with the same values (or based on presence if
+  // the value field is not in the config). Note that this only applies for Thrift transports and/or
+  // protocols that support headers.
+  repeated envoy.api.v2.route.HeaderMatcher headers = 4;
 }
 
 // [#comment:next free field: 2]

source/extensions/filters/network/thrift_proxy/router/BUILD

@@ -40,6 +40,7 @@ envoy_cc_library(
         "//include/envoy/upstream:load_balancer_interface",
         "//include/envoy/upstream:thread_local_cluster_interface",
         "//source/common/common:logger_lib",
+        "//source/common/http:header_utility_lib",
         "//source/common/upstream:load_balancer_lib",
         "//source/extensions/filters/network:well_known_names",
         "//source/extensions/filters/network/thrift_proxy:app_exception_lib",

source/extensions/filters/network/thrift_proxy/router/router_impl.cc

@@ -17,14 +17,22 @@ namespace Router {
 
 RouteEntryImplBase::RouteEntryImplBase(
     const envoy::config::filter::network::thrift_proxy::v2alpha1::Route& route)
-    : cluster_name_(route.route().cluster()) {}
+    : cluster_name_(route.route().cluster()) {
+  for (const auto& header_map : route.match().headers()) {
+    config_headers_.push_back(header_map);
+  }
+}
 
 const std::string& RouteEntryImplBase::clusterName() const { return cluster_name_; }
 
 const RouteEntry* RouteEntryImplBase::routeEntry() const { return this; }
 
 RouteConstSharedPtr RouteEntryImplBase::clusterEntry() const { return shared_from_this(); }
 
+bool RouteEntryImplBase::headersMatch(const Http::HeaderMap& headers) const {
+  return Http::HeaderUtility::matchHeaders(headers, config_headers_);
+}
+
 MethodNameRouteEntryImpl::MethodNameRouteEntryImpl(
     const envoy::config::filter::network::thrift_proxy::v2alpha1::Route& route)
     : RouteEntryImplBase(route), method_name_(route.match().method_name()),
@@ -35,11 +43,13 @@ MethodNameRouteEntryImpl::MethodNameRouteEntryImpl(
 }
 
 RouteConstSharedPtr MethodNameRouteEntryImpl::matches(const MessageMetadata& metadata) const {
-  bool matches =
-      method_name_.empty() || (metadata.hasMethodName() && metadata.methodName() == method_name_);
+  if (RouteEntryImplBase::headersMatch(metadata.headers())) {
+    bool matches =
+        method_name_.empty() || (metadata.hasMethodName() && metadata.methodName() == method_name_);
 
-  if (matches ^ invert_) {
-    return clusterEntry();
+    if (matches ^ invert_) {
+      return clusterEntry();
+    }
   }
 
   return nullptr;
@@ -61,12 +71,14 @@ ServiceNameRouteEntryImpl::ServiceNameRouteEntryImpl(
 }
 
 RouteConstSharedPtr ServiceNameRouteEntryImpl::matches(const MessageMetadata& metadata) const {
-  bool matches = service_name_.empty() ||
-                 (metadata.hasMethodName() &&
-                  StringUtil::startsWith(metadata.methodName().c_str(), service_name_));
+  if (RouteEntryImplBase::headersMatch(metadata.headers())) {
+    bool matches = service_name_.empty() ||
+                   (metadata.hasMethodName() &&
+                    StringUtil::startsWith(metadata.methodName().c_str(), service_name_));
 
-  if (matches ^ invert_) {
-    return clusterEntry();
+    if (matches ^ invert_) {
+      return clusterEntry();
+    }
   }
 
   return nullptr;

source/extensions/filters/network/thrift_proxy/router/router_impl.h

@@ -9,6 +9,7 @@
 #include "envoy/upstream/load_balancer.h"
 
 #include "common/common/logger.h"
+#include "common/http/header_utility.h"
 #include "common/upstream/load_balancer_impl.h"
 
 #include "extensions/filters/network/thrift_proxy/conn_manager.h"
@@ -39,9 +40,11 @@ class RouteEntryImplBase : public RouteEntry,
 
 protected:
   RouteConstSharedPtr clusterEntry() const;
+  bool headersMatch(const Http::HeaderMap& headers) const;
 
 private:
   const std::string cluster_name_;
+  std::vector<Http::HeaderUtility::HeaderData> config_headers_;
 };
 
 typedef std::shared_ptr<const RouteEntryImplBase> RouteEntryImplBaseConstSharedPtr;

test/extensions/filters/network/thrift_proxy/driver/client.py

@@ -79,6 +79,13 @@ def main(cfg, reqhandle, resphandle):
             transport,
             client_type=THeaderTransport.CLIENT_TYPE.HEADER,
         )
+
+        if cfg.headers is not None:
+            pairs = cfg.headers.split(",")
+            for p in pairs:
+                key,value=p.split("=")
+                transport.set_header(key,value)
+
         if cfg.protocol == "binary":
             transport.set_protocol_id(THeaderTransport.T_BINARY_PROTOCOL)
         elif cfg.protocol == "compact":
@@ -159,7 +166,6 @@ def main(cfg, reqhandle, resphandle):
 
     transport.close()
 
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         description="Thrift client tool.",
@@ -225,6 +231,13 @@ def main(cfg, reqhandle, resphandle):
         dest="unix",
         action="store_true",
     )
+    parser.add_argument(
+        "--headers",
+        dest="headers",
+        metavar="KEY=VALUE[,KEY=VALUE]",
+        help="list of comma-delimited, key value pairs to include as tranport headers.",
+    )
+
     cfg = parser.parse_args()
 
     reqhandle = io.BytesIO()

test/extensions/filters/network/thrift_proxy/driver/generate_fixture.sh

@@ -2,12 +2,12 @@
 
 # Generates request and response fixtures for integration tests.
 
-# Usage: generate_fixture.sh <transport> <protocol> [multiplex-service] -- method [param...]
+# Usage: generate_fixture.sh <transport> <protocol> -s [multiplex-service] -H [headers] method [param...]
 
 set -e
 
 function usage() {
-    echo "Usage: $0 <mode> <transport> <protocol> [multiplex-service] -- method [param...]"
+    echo "Usage: $0 <mode> <transport> <protocol> -s [multiplex-service] -H [headers] method [param...]"
     echo "where mode is success, exception, or idl-exception"
     exit 1
 }
@@ -24,24 +24,37 @@ fi
 MODE="$1"
 TRANSPORT="$2"
 PROTOCOL="$3"
-MULTIPLEX="$4"
-if ! shift 4; then
+
+if ! shift 3; then
     usage
 fi
 
-if [[ -z "${MODE}" || -z "${TRANSPORT}" || -z "${PROTOCOL}" || -z "${MULTIPLEX}" ]]; then
+if [[ -z "${MODE}" || -z "${TRANSPORT}" || -z "${PROTOCOL}" ]]; then
     usage
 fi
 
-if [[ "${MULTIPLEX}" != "--" ]]; then
-    if [[ "$1" != "--" ]]; then
-        echo "expected -- after multiplex service name"
-        exit 1
-    fi
-    shift
-else
-    MULTIPLEX=""
-fi
+MULTIPLEX=
+HEADERS=
+while getopts ":s:H:" opt; do
+    case ${opt} in
+        s)
+            MULTIPLEX=$OPTARG
+            ;;
+        H)
+            HEADERS=$OPTARG
+            ;;
+
+        \?)
+            echo "Invalid Option: -$OPTARG" >&2
+            exit 1
+            ;;
+        :)
+            echo "Invalid Option: -$OPTARG requires an argument" >&2
+            exit 1
+            ;;
+    esac
+done
+shift $((OPTIND -1))
 
 METHOD="$1"
 if [[ "${METHOD}" == "" ]]; then
@@ -59,8 +72,8 @@ SERVICE_FLAGS=("--addr" "${SOCKET}"
                "--protocol" "${PROTOCOL}")
 
 if [[ -n "$MULTIPLEX" ]]; then
-    SERVICE_FLAGS[9]="--multiplex"
-    SERVICE_FLAGS[10]="${MULTIPLEX}"
+    SERVICE_FLAGS+=("--multiplex")
+    SERVICE_FLAGS+=("${MULTIPLEX}")
 
     REQUEST_FILE="${FIXTURE_DIR}/${TRANSPORT}-${PROTOCOL}-${MULTIPLEX}-${MODE}.request"
     RESPONSE_FILE="${FIXTURE_DIR}/${TRANSPORT}-${PROTOCOL}-${MULTIPLEX}-${MODE}.response"
@@ -84,6 +97,11 @@ while [[ ! -a "${SOCKET}" ]]; do
     fi
 done
 
+if [[ -n "$HEADERS" ]]; then
+    SERVICE_FLAGS+=("--headers")
+    SERVICE_FLAGS+=("$HEADERS")
+fi
+
 "${DRIVER_DIR}/client" "${SERVICE_FLAGS[@]}" \
                        --request "${REQUEST_FILE}" \
                        --response "${RESPONSE_FILE}" \

test/extensions/filters/network/thrift_proxy/integration.cc

@@ -63,9 +63,21 @@ void BaseThriftIntegrationTest::preparePayloads(const PayloadOptions& options,
   };
 
   if (options.service_name_) {
+    args.push_back("-s");
     args.push_back(*options.service_name_);
   }
-  args.push_back("--");
+
+  if (options.headers_.size() > 0) {
+    args.push_back("-H");
+
+    std::vector<std::string> headers;
+    std::transform(options.headers_.begin(), options.headers_.end(), std::back_inserter(headers),
+                   [](const std::pair<std::string, std::string>& header) -> std::string {
+                     return header.first + "=" + header.second;
+                   });
+    args.push_back(StringUtil::join(headers, ","));
+  }
+
   args.push_back(options.method_name_);
   std::copy(options.method_args_.begin(), options.method_args_.end(), std::back_inserter(args));
 

test/extensions/filters/network/thrift_proxy/integration.h

@@ -29,9 +29,10 @@ enum class DriverMode {
 struct PayloadOptions {
   PayloadOptions(TransportType transport, ProtocolType protocol, DriverMode mode,
                  absl::optional<std::string> service_name, std::string method_name,
-                 std::vector<std::string> method_args = {})
+                 std::vector<std::string> method_args = {},
+                 std::vector<std::pair<std::string, std::string>> headers = {})
       : transport_(transport), protocol_(protocol), mode_(mode), service_name_(service_name),
-        method_name_(method_name), method_args_(method_args) {}
+        method_name_(method_name), method_args_(method_args), headers_(headers) {}
 
   std::string modeName() const;
   std::string transportName() const;
@@ -43,6 +44,7 @@ struct PayloadOptions {
   const absl::optional<std::string> service_name_;
   const std::string method_name_;
   const std::vector<std::string> method_args_;
+  const std::vector<std::pair<std::string, std::string>> headers_;
 };
 
 class BaseThriftIntegrationTest : public BaseIntegrationTest {

test/extensions/filters/network/thrift_proxy/integration_test.cc

@@ -34,12 +34,29 @@ class ThriftConnManagerIntegrationTest
                     cluster: "cluster_0"
                 - match:
                     method_name: "execute"
+                    headers:
+                    - name: "x-header-1"
+                      exact_match: "x-value-1"
+                    - name: "x-header-2"
+                      regex_match: "0.[5-9]"
+                    - name: "x-header-3"
+                      range_match:
+                        start: 100
+                        end: 200
+                    - name: "x-header-4"
+                      prefix_match: "user_id:"
+                    - name: "x-header-5"
+                      suffix_match: "asdf"
                   route:
                     cluster: "cluster_1"
                 - match:
-                    method_name: "poke"
+                    method_name: "execute"
                   route:
                     cluster: "cluster_2"
+                - match:
+                    method_name: "poke"
+                  route:
+                    cluster: "cluster_3"
       )EOF";
   }
 
@@ -51,7 +68,16 @@ class ThriftConnManagerIntegrationTest
       service_name = "svcname";
     }
 
-    PayloadOptions options(transport_, protocol_, mode, service_name, "execute");
+    std::vector<std::pair<std::string, std::string>> headers;
+    if (transport_ == TransportType::Header) {
+      headers.push_back(std::make_pair("x-header-1", "x-value-1"));
+      headers.push_back(std::make_pair("x-header-2", "0.6"));
+      headers.push_back(std::make_pair("x-header-3", "150"));
+      headers.push_back(std::make_pair("x-header-4", "user_id:10"));
+      headers.push_back(std::make_pair("x-header-5", "garbage_asdf"));
+    }
+
+    PayloadOptions options(transport_, protocol_, mode, service_name, "execute", {}, headers);
     preparePayloads(options, request_bytes_, response_bytes_);
     ASSERT(request_bytes_.length() > 0);
     ASSERT(response_bytes_.length() > 0);
@@ -76,16 +102,14 @@ class ThriftConnManagerIntegrationTest
   // We allocate as many upstreams as there are clusters, with each upstream being allocated
   // to clusters in the order they're defined in the bootstrap config.
   void initializeCommon() {
-    setUpstreamCount(3);
+    setUpstreamCount(4);
 
     config_helper_.addConfigModifier([](envoy::config::bootstrap::v2::Bootstrap& bootstrap) {
-      auto* c1 = bootstrap.mutable_static_resources()->add_clusters();
-      c1->MergeFrom(bootstrap.static_resources().clusters()[0]);
-      c1->set_name("cluster_1");
-
-      auto* c2 = bootstrap.mutable_static_resources()->add_clusters();
-      c2->MergeFrom(bootstrap.static_resources().clusters()[0]);
-      c2->set_name("cluster_2");
+      for (int i = 1; i < 4; i++) {
+        auto* c = bootstrap.mutable_static_resources()->add_clusters();
+        c->MergeFrom(bootstrap.static_resources().clusters()[0]);
+        c->set_name(fmt::format("cluster_{}", i));
+      }
     });
 
     BaseThriftIntegrationTest::initialize();
@@ -101,11 +125,13 @@ class ThriftConnManagerIntegrationTest
   // while oneway's are handled by the "poke" method. All other requests
   // are handled by "execute".
   FakeUpstream* getExpectedUpstream(bool oneway) {
-    int upstreamIdx = 1;
+    int upstreamIdx = 2;
     if (multiplexed_) {
       upstreamIdx = 0;
     } else if (oneway) {
-      upstreamIdx = 2;
+      upstreamIdx = 3;
+    } else if (transport_ == TransportType::Header) {
+      upstreamIdx = 1;
     }
 
     return fake_upstreams_[upstreamIdx].get();