rocketmq_proxy: implement rocketmq proxy

CODEOWNERS

@@ -16,6 +16,8 @@
 extensions/filters/common/original_src @snowp @klarose
 # dubbo_proxy extension
 /*/extensions/filters/network/dubbo_proxy @zyfjeff @lizan
+# rocketmq_proxy extension
+/*/extensions/filters/network/rocketmq_proxy @aaron-ai @lizhanhui @lizan
 # thrift_proxy extension
 /*/extensions/filters/network/thrift_proxy @zuercher @brian-pane
 # compressor used by http compression filters

api/BUILD

@@ -215,6 +215,7 @@ proto_library(
         "//envoy/extensions/filters/network/ratelimit/v3:pkg",
         "//envoy/extensions/filters/network/rbac/v3:pkg",
         "//envoy/extensions/filters/network/redis_proxy/v3:pkg",
+        "//envoy/extensions/filters/network/rocketmq_proxy/v3:pkg",
         "//envoy/extensions/filters/network/sni_cluster/v3:pkg",
         "//envoy/extensions/filters/network/sni_dynamic_forward_proxy/v3alpha:pkg",
         "//envoy/extensions/filters/network/tcp_proxy/v3:pkg",

api/envoy/extensions/filters/network/rocketmq_proxy/v3/BUILD

@@ -0,0 +1,14 @@
+# DO NOT EDIT. This file is generated by tools/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = [
+        "//envoy/config/core/v3:pkg",
+        "//envoy/config/route/v3:pkg",
+        "//envoy/type/matcher/v3:pkg",
+        "@com_github_cncf_udpa//udpa/annotations:pkg",
+    ],
+)

api/envoy/extensions/filters/network/rocketmq_proxy/v3/README.md

@@ -0,0 +1 @@
+Protocol buffer definitions for the Rocketmq proxy.

api/envoy/extensions/filters/network/rocketmq_proxy/v3/rocketmq_proxy.proto

@@ -0,0 +1,36 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.network.rocketmq_proxy.v3;
+
+import "envoy/extensions/filters/network/rocketmq_proxy/v3/route.proto";
+
+import "google/protobuf/any.proto";
+import "google/protobuf/duration.proto";
+
+import "udpa/annotations/status.proto";
+import "udpa/annotations/versioning.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.network.rocketmq_proxy.v3";
+option java_outer_classname = "RocketmqProxyProto";
+option java_multiple_files = true;
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: RocketMQ Proxy]
+// RocketMQ Proxy :ref:`configuration overview <config_network_filters_rocketmq_proxy>`.
+// [#extension: envoy.filters.network.rocketmq_proxy]
+
+message RocketmqProxy {
+  // The human readable prefix to use when emitting statistics.
+  string stat_prefix = 1 [(validate.rules).string = {min_bytes: 1}];
+
+  // The route table for the connection manager is specified in this property.
+  RouteConfiguration route_config = 2;
+
+  // The largest duration transient object expected to live, more than 10s is recommended.
+  google.protobuf.Duration transient_object_life_span = 3;
+
+  // If develop_mode is enabled, this proxy plugin may work without dedicated traffic intercepting
+  // facility without considering backward compatibility of exiting RocketMQ client SDK.
+  bool develop_mode = 4;
+}

api/envoy/extensions/filters/network/rocketmq_proxy/v3/route.proto

@@ -0,0 +1,55 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.network.rocketmq_proxy.v3;
+
+import "envoy/config/core/v3/base.proto";
+import "envoy/config/route/v3/route_components.proto";
+import "envoy/type/matcher/v3/string.proto";
+
+import "udpa/annotations/status.proto";
+import "udpa/annotations/versioning.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.network.rocketmq_proxy.v3";
+option java_outer_classname = "RouteProto";
+option java_multiple_files = true;
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Rocketmq Proxy Route Configuration]
+// Rocketmq Proxy :ref:`configuration overview <config_network_filters_rocketmq_proxy>`.
+
+message RouteConfiguration {
+  // The name of the route configuration.
+  string name = 1;
+
+  // The list of routes that will be matched, in order, against incoming requests. The first route
+  // that matches will be used.
+  repeated Route routes = 2;
+}
+
+message Route {
+  // Route matching parameters.
+  RouteMatch match = 1 [(validate.rules).message = {required: true}];
+
+  // Route request to some upstream cluster.
+  RouteAction route = 2 [(validate.rules).message = {required: true}];
+}
+
+message RouteMatch {
+  // The name of the topic.
+  type.matcher.v3.StringMatcher topic = 1 [(validate.rules).message = {required: true}];
+
+  // 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).
+  repeated config.route.v3.HeaderMatcher headers = 2;
+}
+
+message RouteAction {
+  // Indicates the upstream cluster to which the request should be routed.
+  string cluster = 1 [(validate.rules).string = {min_bytes: 1}];
+
+  // Optional endpoint metadata match criteria used by the subset load balancer.
+  config.core.v3.Metadata metadata_match = 2;
+}

api/versioning/BUILD

@@ -96,6 +96,7 @@ proto_library(
         "//envoy/extensions/filters/network/ratelimit/v3:pkg",
         "//envoy/extensions/filters/network/rbac/v3:pkg",
         "//envoy/extensions/filters/network/redis_proxy/v3:pkg",
+        "//envoy/extensions/filters/network/rocketmq_proxy/v3:pkg",
         "//envoy/extensions/filters/network/sni_cluster/v3:pkg",
         "//envoy/extensions/filters/network/sni_dynamic_forward_proxy/v3alpha:pkg",
         "//envoy/extensions/filters/network/tcp_proxy/v3:pkg",

docs/root/configuration/listeners/network_filters/network_filters.rst

@@ -23,6 +23,7 @@ filters.
   rate_limit_filter
   rbac_filter
   redis_proxy_filter
+  rocketmq_proxy_filter
   tcp_proxy_filter
   thrift_proxy_filter
   sni_cluster_filter

docs/root/configuration/listeners/network_filters/rocketmq_proxy_filter.rst

@@ -0,0 +1,76 @@
+.. _config_network_filters_rocketmq_proxy:
+
+RocketMQ proxy
+==============
+
+Apache RocketMQ is a distributed messaging system, which is composed of four types of roles: producer, consumer, name
+server and broker server. The former two are embedded into user application in form of SDK; whilst the latter are
+standalone servers.
+
+A message in RocketMQ carries a topic as its destination and optionally one or more tags as application specific labels.
+
+Producers are used to send messages to brokers according to their topics. Similar to many distributed systems,
+producers need to know how to connect to these serving brokers. To achieve this goal, RocketMQ provides name server
+clusters for producers to lookup. Namely, when producers attempts to send messages with a new topic, it first
+tries to lookup the addresses(called route info) of brokers that serve the topic from name servers. Once producers
+get the route info of the topic, they actively cache them in memory and renew them periodically thereafter. This
+mechanism, though simple, effectively keeps service availability high without demanding availability of name server
+service.
+
+Brokers provides messaging service to end users. In addition to various messaging services, they also periodically
+report health status and route info of topics currently served to name servers.
+
+Major role of the name server is to serve querying of route info  for a topic. Additionally, it also purges route info
+entries once the belonging brokers fail to report their health info for a configured period of time. This ensures
+clients almost always connect to brokers that are online and ready to serve.
+
+Consumers are used by application to pull message from brokers. They perform similar heartbeats to maintain alive
+status. RocketMQ brokers support two message-fetch approaches: long-pulling and pop.
+
+Using the first approach, consumers have to implement load-balancing algorithm. The pop approach, in the perspective of
+consumers, is stateless.
+
+Envoy RocketMQ filter proxies requests and responses between producers/consumer and brokers. Various statistical items
+are collected to enhance observability.
+
+At present, pop-based message fetching is implemented. Long-pulling will be implemented in the next pull request.
+
+.. _config_network_filters_rocketmq_proxy_stats:
+
+Statistics
+----------
+
+Every configured rocketmq proxy filter has statistics rooted at *rocketmq.<stat_prefix>.* with the
+following statistics:
+
+.. csv-table::
+  :header: Name, Type, Description
+  :widths: 1, 1, 2
+
+  request, Counter, Total requests
+  request_decoding_error, Counter, Total decoding error requests
+  request_decoding_success, Counter, Total decoding success requests
+  response, Counter, Total responses
+  response_decoding_error, Counter, Total decoding error responses
+  response_decoding_success, Counter, Total decoding success responses
+  response_error, Counter, Total error responses
+  response_success, Counter, Total success responses
+  heartbeat, Counter, Total heartbeat requests
+  unregister, Counter, Total unregister requests
+  get_topic_route, Counter, Total getting topic route requests
+  send_message_v1, Counter, Total sending message v1 requests
+  send_message_v2, Counter, Total sending message v2 requests
+  pop_message, Counter, Total poping message requests
+  ack_message, Counter, Total acking message requests
+  get_consumer_list, Counter, Total getting consumer list requests
+  maintenance_failure, Counter, Total maintenance failure
+  request_active, Gauge, Total active requests
+  send_message_v1_active, Gauge, Total active sending message v1 requests
+  send_message_v2_active, Gauge, Total active sending message v2 requests
+  pop_message_active, Gauge, Total active poping message active requests
+  get_topic_route_active, Gauge, Total active geting topic route requests
+  send_message_pending, Gauge, Total pending sending message requests
+  pop_message_pending, Gauge, Total pending poping message requests
+  get_topic_route_pending, Gauge, Total pending geting topic route requests
+  total_pending, Gauge, Total pending requests
+  request_time_ms, Histogram, Request time in milliseconds

docs/root/version_history/current.rst

@@ -17,6 +17,7 @@ Changes
   Can be reverted temporarily by setting runtime feature `envoy.reloadable_features.fix_upgrade_response` to false.
 * logger: added :ref:`--log-format-prefix-with-location <operations_cli>` command line option to prefix '%v' with file path and line number.
 * network filters: added a :ref:`postgres proxy filter <config_network_filters_postgres_proxy>`.
+* network filters: added a :ref:`rocketmq proxy filter <config_network_filters_rocketmq_proxy>`.
 * request_id: added to :ref:`always_set_request_id_in_response setting <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.always_set_request_id_in_response>`
   to set :ref:`x-request-id <config_http_conn_man_headers_x-request-id>` header in response even if
   tracing is not forced.

generated_api_shadow/BUILD

@@ -77,6 +77,7 @@ proto_library(
         "//envoy/config/filter/network/rate_limit/v2:pkg",
         "//envoy/config/filter/network/rbac/v2:pkg",
         "//envoy/config/filter/network/redis_proxy/v2:pkg",
+        "//envoy/config/filter/network/rocketmq_proxy/v3:pkg",
         "//envoy/config/filter/network/sni_cluster/v2:pkg",
         "//envoy/config/filter/network/tcp_proxy/v2:pkg",
         "//envoy/config/filter/network/thrift_proxy/v2alpha1:pkg",

source/common/common/logger.h

@@ -34,6 +34,7 @@ namespace Logger {
   FUNCTION(conn_handler)                                                                           \
   FUNCTION(decompression)                                                                          \
   FUNCTION(dubbo)                                                                                  \
+  FUNCTION(rocketmq)                                                                               \
   FUNCTION(file)                                                                                   \
   FUNCTION(filter)                                                                                 \
   FUNCTION(forward_proxy)                                                                          \

source/extensions/extensions_build_config.bzl

@@ -96,6 +96,7 @@ EXTENSIONS = {
     "envoy.filters.network.ratelimit":                  "//source/extensions/filters/network/ratelimit:config",
     "envoy.filters.network.rbac":                       "//source/extensions/filters/network/rbac:config",
     "envoy.filters.network.redis_proxy":                "//source/extensions/filters/network/redis_proxy:config",
+    "envoy.filters.network.rocketmq_proxy":             "//source/extensions/filters/network/rocketmq_proxy:config",
     "envoy.filters.network.tcp_proxy":                  "//source/extensions/filters/network/tcp_proxy:config",
     "envoy.filters.network.thrift_proxy":               "//source/extensions/filters/network/thrift_proxy:config",
     "envoy.filters.network.sni_cluster":                "//source/extensions/filters/network/sni_cluster:config",