Merge branch 'main' into protocol_integration_test

Signed-off-by: Ryan Hamilton <[email protected]>

api/BUILD

@@ -83,6 +83,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/network/client_ssl_auth/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/action/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/codecs/dubbo/v3:pkg",
+        "//contrib/envoy/extensions/filters/network/generic_proxy/codecs/http1/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/codecs/kafka/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/matcher/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/router/v3:pkg",

api/contrib/envoy/extensions/filters/network/generic_proxy/codecs/http1/v3/BUILD

@@ -0,0 +1,12 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = [
+        "@com_github_cncf_xds//udpa/annotations:pkg",
+        "@com_github_cncf_xds//xds/annotations/v3:pkg",
+    ],
+)

api/contrib/envoy/extensions/filters/network/generic_proxy/codecs/http1/v3/http1.proto

@@ -0,0 +1,49 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.network.generic_proxy.codecs.http1.v3;
+
+import "google/protobuf/wrappers.proto";
+
+import "xds/annotations/v3/status.proto";
+
+import "udpa/annotations/status.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.network.generic_proxy.codecs.http1.v3";
+option java_outer_classname = "Http1Proto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/contrib/envoy/extensions/filters/network/generic_proxy/codecs/http1/v3;http1v3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+option (xds.annotations.v3.file_status).work_in_progress = true;
+
+// [#protodoc-title: HTTP1 codec configuration for Generic Proxy]
+// [#extension: envoy.generic_proxy.codecs.http1]
+
+// Configuration for HTTP codec. This HTTP1 codec is used to parse and serialize HTTP1 messages
+// for the generic proxy filter.
+// Any decoding error will result in the generic proxy closing the connection.
+//
+// .. note::
+//   This codec only supports HTTP1.1 messages and does not support HTTP1.0 messages. And it limits
+//   part of the HTTP1.1 features, such as upgrade, connect, etc.
+//   This codec is mainly designed for the features evaluation of the generic proxy filter. Please
+//   be cautious when using it in production.
+message Http1CodecConfig {
+  // If true, the codec will parse and serialize HTTP1 messages in a single frame per message.
+  //
+  // A frame is a minimal unit of data that can be processed by the generic proxy. If false, the
+  // codec will parse and serialize HTTP1 messages in a streaming way. In this case, the codec
+  // will output multiple frames for a single HTTP1 message to the generic proxy.
+  // If true, the codec will buffer the entire HTTP1 message body before sending it to the generic
+  // proxy. This may have better performance in small message scenarios and is more friendly to
+  // handle the HTTP1 message body. This also may result in higher memory usage and latency if
+  // the message body is large.
+  //
+  // Default is true.
+  google.protobuf.BoolValue single_frame_mode = 1;
+
+  // The maximum size of the HTTP1 message body in bytes. If not set, 8*1024*1024 (8MB) is used.
+  // This only makes sense when single_frame_mode is true.
+  // If the HTTP1 message body size exceeds this value, this will result in a decoding error
+  // and the generic proxy will close the connection.
+  google.protobuf.UInt32Value max_buffer_size = 2;
+}

api/envoy/extensions/access_loggers/fluentd/v3/BUILD

@@ -5,5 +5,8 @@ load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
 licenses(["notice"])  # Apache 2
 
 api_proto_package(
-    deps = ["@com_github_cncf_xds//udpa/annotations:pkg"],
+    deps = [
+        "//envoy/config/core/v3:pkg",
+        "@com_github_cncf_xds//udpa/annotations:pkg",
+    ],
 )

api/envoy/extensions/access_loggers/fluentd/v3/fluentd.proto

@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package envoy.extensions.access_loggers.fluentd.v3;
 
+import "envoy/config/core/v3/backoff.proto";
+
 import "google/protobuf/duration.proto";
 import "google/protobuf/struct.proto";
 import "google/protobuf/wrappers.proto";
@@ -22,8 +24,19 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // the Fluentd Forward Protocol as described in: `Fluentd Forward Protocol Specification
 // <https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1>`_.
 // [#extension: envoy.access_loggers.fluentd]
-// [#next-free-field: 7]
+// [#next-free-field: 8]
 message FluentdAccessLogConfig {
+  message RetryOptions {
+    // The number of times the logger will attempt to connect to the upstream during reconnects.
+    // By default, there is no limit. The logger will attempt to reconnect to the upstream each time
+    // connecting to the upstream failed or the upstream connection had been closed for any reason.
+    google.protobuf.UInt32Value max_connect_attempts = 1;
+
+    // Sets the backoff strategy. If this value is not set, the default base backoff interval is 500
+    // milliseconds and the default max backoff interval is 5 seconds (10 times the base interval).
+    config.core.v3.BackoffStrategy backoff_options = 2;
+  }
+
   // The upstream cluster to connect to for streaming the Fluentd messages.
   string cluster = 1 [(validate.rules).string = {min_len: 1}];
 
@@ -67,4 +80,9 @@ message FluentdAccessLogConfig {
   //    "message": "My error message"
   //  }
   google.protobuf.Struct record = 6 [(validate.rules).message = {required: true}];
+
+  // Optional retry, in case upstream connection has failed. If this field is not set, the default values will be applied,
+  // as specified in the :ref:`RetryOptions <envoy_v3_api_msg_extensions.access_loggers.fluentd.v3.FluentdAccessLogConfig.RetryOptions>`
+  // configuration.
+  RetryOptions retry_options = 7;
 }

api/envoy/extensions/filters/http/rbac/v3/rbac.proto

@@ -22,7 +22,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // [#extension: envoy.filters.http.rbac]
 
 // RBAC filter config.
-// [#next-free-field: 6]
+// [#next-free-field: 7]
 message RBAC {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.http.rbac.v2.RBAC";
@@ -34,6 +34,11 @@ message RBAC {
   config.rbac.v3.RBAC rules = 1
       [(udpa.annotations.field_migrate).oneof_promotion = "rules_specifier"];
 
+  // If specified, rules will emit stats with the given prefix.
+  // This is useful to distinguish the stat when there are more than 1 RBAC filter configured with
+  // rules.
+  string rules_stat_prefix = 6;
+
   // The match tree to use when resolving RBAC action for incoming requests. Requests do not
   // match any matcher will be denied.
   // If absent, no enforcing RBAC matcher will be applied.

api/versioning/BUILD

@@ -21,6 +21,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/network/client_ssl_auth/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/action/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/codecs/dubbo/v3:pkg",
+        "//contrib/envoy/extensions/filters/network/generic_proxy/codecs/http1/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/codecs/kafka/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/matcher/v3:pkg",
         "//contrib/envoy/extensions/filters/network/generic_proxy/router/v3:pkg",

changelogs/current.yaml

@@ -313,6 +313,11 @@ new_features:
     added a :ref:`configuration option
     <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.append_local_overload>` to add
     ``x-envoy-local-overloaded`` header when Overload Manager is triggered.
+- area: access_loggers
+  change: |
+    Added :ref:`retry options
+    <envoy_v3_api_msg_extensions.access_loggers.fluentd.v3.FluentdAccessLogConfig.RetryOptions>` to Fluentd Access Logger to
+    support upstream reconnect options, backoff intervals.
 - area: tracing
   change: |
     Added support to configure a Dynatrace sampler for the OpenTelemetry tracer.
@@ -339,6 +344,10 @@ new_features:
     Update ``aws_request_signing`` filter to support optionally sending the aws signature in query parameters rather than headers,
     by specifying the :ref:`query_string <envoy_v3_api_field_extensions.filters.http.aws_request_signing.v3.AwsRequestSigning.query_string>`
     configuration section.
+- area: rbac
+  change: |
+    Added :ref:`rules_stat_prefix <envoy_v3_api_field_extensions.filters.http.rbac.v3.RBAC.rules_stat_prefix>`
+    to allow adding custom prefix to the stats emitted by rules.
 
 deprecated:
 - area: listener

contrib/contrib_build_config.bzl

@@ -73,6 +73,7 @@ CONTRIB_EXTENSIONS = {
     #
     "envoy.filters.generic.router":                             "//contrib/generic_proxy/filters/network/source/router:config",
     "envoy.generic_proxy.codecs.dubbo":                         "//contrib/generic_proxy/filters/network/source/codecs/dubbo:config",
+    "envoy.generic_proxy.codecs.http1":                         "//contrib/generic_proxy/filters/network/source/codecs/http1:config",
     "envoy.generic_proxy.codecs.kafka":                         "//contrib/generic_proxy/filters/network/source/codecs/kafka:config",
 
     #

contrib/extensions_metadata.yaml

@@ -144,6 +144,13 @@ envoy.generic_proxy.codecs.kafka:
   status: wip
   type_urls:
   - envoy.extensions.filters.network.generic_proxy.codecs.kafka.v3.KafkaCodecConfig
+envoy.generic_proxy.codecs.http1:
+  categories:
+  - envoy.generic_proxy.codecs
+  security_posture: requires_trusted_downstream_and_upstream
+  status: wip
+  type_urls:
+  - envoy.extensions.filters.network.generic_proxy.codecs.http1.v3.Http1CodecConfig
 envoy.router.cluster_specifier_plugin.golang:
   categories:
   - envoy.router.cluster_specifier_plugin

contrib/generic_proxy/filters/network/source/codecs/dubbo/config.cc

@@ -4,7 +4,7 @@
 
 #include "envoy/registry/registry.h"
 
-#include "source/extensions/common/dubbo/message_impl.h"
+#include "source/extensions/common/dubbo/message.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -31,20 +31,9 @@ Common::Dubbo::ResponseStatus genericStatusToStatus(StatusCode code) {
 } // namespace
 
 void DubboRequest::forEach(IterateCallback callback) const {
-  const auto* typed_request =
-      dynamic_cast<Common::Dubbo::RpcRequestImpl*>(&inner_metadata_->mutableRequest());
-  ASSERT(typed_request != nullptr);
-
-  for (const auto& pair : typed_request->attachment().attachment()) {
-    ASSERT(pair.first != nullptr && pair.second != nullptr);
-
-    if (pair.first->type() == Hessian2::Object::Type::String &&
-        pair.second->type() == Hessian2::Object::Type::String) {
-      ASSERT(pair.first->toString().has_value() && pair.second->toString().has_value());
-
-      if (!callback(pair.first->toString().value().get(), pair.second->toString().value().get())) {
-        break;
-      }
+  for (const auto& [key, val] : inner_metadata_->request().content().attachments()) {
+    if (!callback(key, val)) {
+      break;
     }
   }
 }
@@ -53,27 +42,21 @@ absl::optional<absl::string_view> DubboRequest::get(absl::string_view key) const
   if (key == VERSION_KEY) {
     return inner_metadata_->request().serviceVersion();
   }
-  const auto* typed_request =
-      dynamic_cast<Common::Dubbo::RpcRequestImpl*>(&inner_metadata_->mutableRequest());
-  ASSERT(typed_request != nullptr);
 
-  return typed_request->attachment().lookup(key);
+  auto it = inner_metadata_->request().content().attachments().find(key);
+  if (it == inner_metadata_->request().content().attachments().end()) {
+    return absl::nullopt;
+  }
+
+  return absl::string_view{it->second};
 }
 
 void DubboRequest::set(absl::string_view key, absl::string_view val) {
-  auto* typed_request =
-      dynamic_cast<Common::Dubbo::RpcRequestImpl*>(&inner_metadata_->mutableRequest());
-  ASSERT(typed_request != nullptr);
-
-  typed_request->mutableAttachment()->insert(key, val);
+  inner_metadata_->request().content().setAttachment(key, val);
 }
 
 void DubboRequest::erase(absl::string_view key) {
-  auto* typed_request =
-      dynamic_cast<Common::Dubbo::RpcRequestImpl*>(&inner_metadata_->mutableRequest());
-  ASSERT(typed_request != nullptr);
-
-  typed_request->mutableAttachment()->remove(key);
+  inner_metadata_->request().content().delAttachment(key);
 }
 
 void DubboResponse::refreshStatus() {
@@ -85,45 +68,48 @@ void DubboResponse::refreshStatus() {
   const auto status = inner_metadata_->context().responseStatus();
   const auto optional_type = inner_metadata_->response().responseType();
 
-  // The final status is not ok if the response status is not ResponseStatus::Ok
-  // anyway.
-  bool response_ok = (status == Common::Dubbo::ResponseStatus::Ok);
+  if (status != Common::Dubbo::ResponseStatus::Ok) {
+    status_ = StreamStatus(static_cast<uint8_t>(status), false);
+    return;
+  }
 
+  bool response_ok = true;
   // The final status is not ok if the response type is ResponseWithException or
   // ResponseWithExceptionWithAttachments even if the response status is Ok.
-  if (status == Common::Dubbo::ResponseStatus::Ok) {
-    ASSERT(optional_type.has_value());
-    auto type = optional_type.value_or(RpcResponseType::ResponseWithException);
-    if (type == RpcResponseType::ResponseWithException ||
-        type == RpcResponseType::ResponseWithExceptionWithAttachments) {
-      response_ok = false;
-    }
+  ASSERT(optional_type.has_value());
+  auto type = optional_type.value_or(RpcResponseType::ResponseWithException);
+  if (type == RpcResponseType::ResponseWithException ||
+      type == RpcResponseType::ResponseWithExceptionWithAttachments) {
+    response_ok = false;
   }
 
   status_ = StreamStatus(static_cast<uint8_t>(status), response_ok);
 }
 
 DubboCodecBase::DubboCodecBase(Common::Dubbo::DubboCodecPtr codec) : codec_(std::move(codec)) {}
 
-ResponsePtr DubboServerCodec::respond(Status status, absl::string_view,
+ResponsePtr DubboServerCodec::respond(Status status, absl::string_view data,
                                       const Request& origin_request) {
   const auto* typed_request = dynamic_cast<const DubboRequest*>(&origin_request);
   ASSERT(typed_request != nullptr);
 
   Common::Dubbo::ResponseStatus response_status = genericStatusToStatus(status.code());
 
   absl::optional<Common::Dubbo::RpcResponseType> optional_type;
-  absl::string_view content;
 
   if (response_status == Common::Dubbo::ResponseStatus::Ok) {
     optional_type.emplace(Common::Dubbo::RpcResponseType::ResponseWithException);
-    content = "exception_via_proxy";
+  }
+  auto response = Common::Dubbo::DirectResponseUtil::localResponse(
+      *typed_request->inner_metadata_, response_status, optional_type, data);
+
+  if (!status.ok()) {
+    response->mutableResponse().content().setAttachment("reason", status.message());
   } else {
-    content = status.message();
+    response->mutableResponse().content().setAttachment("reason", "envoy_response");
   }
 
-  return std::make_unique<DubboResponse>(Common::Dubbo::DirectResponseUtil::localResponse(
-      *typed_request->inner_metadata_, response_status, optional_type, content));
+  return std::make_unique<DubboResponse>(std::move(response));
 }
 
 CodecFactoryPtr