Kafka-mesh filter

api/BUILD

@@ -60,6 +60,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/http/squash/v3:pkg",
         "//contrib/envoy/extensions/filters/http/sxg/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/kafka_broker/v3:pkg",
+        "//contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/mysql_proxy/v3:pkg",
         "//contrib/envoy/extensions/filters/network/postgres_proxy/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/rocketmq_proxy/v3:pkg",

api/contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha/BUILD

@@ -0,0 +1,9 @@
+# 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_udpa//udpa/annotations:pkg"],
+)

api/contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha/kafka_mesh.proto

@@ -0,0 +1,58 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.network.kafka_mesh.v3alpha;
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.network.kafka_mesh.v3alpha";
+option java_outer_classname = "KafkaMeshProto";
+option java_multiple_files = true;
+option (udpa.annotations.file_status).work_in_progress = true;
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Kafka Mesh]
+// Kafka Mesh :ref:`configuration overview <config_network_filters_kafka_mesh>`.
+// [#extension: envoy.filters.network.kafka_mesh]
+
+message KafkaMesh {
+  // Envoy's host that's advertised to clients.
+  // Has the same meaning as corresponding Kafka broker properties.
+  // Usually equal to filter chain's listener config, but needs to be reachable by clients
+  // (so 0.0.0.0 will not work).
+  string advertised_host = 1 [(validate.rules).string = {min_len: 1}];
+
+  // Envoy's port that's advertised to clients.
+  int32 advertised_port = 2 [(validate.rules).int32 = {gt: 0}];
+
+  // Upstream clusters this filter will connect to.
+  repeated KafkaClusterDefinition upstream_clusters = 3;
+
+  // Rules that will decide which cluster gets which request.
+  repeated ForwardingRule forwarding_rules = 4;
+}
+
+message KafkaClusterDefinition {
+  // Cluster name.
+  string cluster_name = 1 [(validate.rules).string = {min_len: 1}];
+
+  // Kafka cluster address.
+  string bootstrap_servers = 2 [(validate.rules).string = {min_len: 1}];
+
+  // Default number of partitions present in this cluster.
+  // This is especially important for clients that do not specify partition in their payloads and depend on this value for hashing.
+  int32 partition_count = 3 [(validate.rules).int32 = {gt: 0}];
+
+  // Custom configuration passed to Kafka producer.
+  map<string, string> producer_config = 4;
+}
+
+message ForwardingRule {
+  // Cluster name.
+  string target_cluster = 1;
+
+  oneof trigger {
+    // Intended place for future types of forwarding rules.
+    string topic_prefix = 2;
+  }
+}

api/versioning/BUILD

@@ -12,6 +12,7 @@ proto_library(
         "//contrib/envoy/extensions/filters/http/squash/v3:pkg",
         "//contrib/envoy/extensions/filters/http/sxg/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/kafka_broker/v3:pkg",
+        "//contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/mysql_proxy/v3:pkg",
         "//contrib/envoy/extensions/filters/network/postgres_proxy/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/network/rocketmq_proxy/v3:pkg",

bazel/BUILD

@@ -586,3 +586,8 @@ alias(
     name = "remote_jdk11",
     actual = "@bazel_tools//tools/jdk:remote_jdk11",
 )
+
+alias(
+    name = "windows",
+    actual = "@bazel_tools//src/conditions:windows",
+)

bazel/repository_locations.bzl

@@ -935,7 +935,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         strip_prefix = "kafka-{version}/clients/src/main/resources/common/message",
         urls = ["https://github.com/apache/kafka/archive/{version}.zip"],
         use_category = ["dataplane_ext"],
-        extensions = ["envoy.filters.network.kafka_broker"],
+        extensions = ["envoy.filters.network.kafka_broker", "envoy.filters.network.kafka_mesh"],
         release_date = "2020-03-03",
         cpe = "cpe:2.3:a:apache:kafka:*",
     ),
@@ -948,7 +948,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         strip_prefix = "librdkafka-{version}",
         urls = ["https://github.com/edenhill/librdkafka/archive/v{version}.tar.gz"],
         use_category = ["dataplane_ext"],
-        extensions = ["envoy.filters.network.kafka_broker"],
+        extensions = ["envoy.filters.network.kafka_mesh"],
         release_date = "2021-05-10",
         cpe = "N/A",
     ),

contrib/contrib_build_config.bzl

@@ -12,6 +12,7 @@ CONTRIB_EXTENSIONS = {
     #
 
     "envoy.filters.network.kafka_broker":                       "//contrib/kafka/filters/network/source:kafka_broker_config_lib",
+    "envoy.filters.network.kafka_mesh":                         "//contrib/kafka/filters/network/source/mesh:config_lib",
     "envoy.filters.network.mysql_proxy":                        "//contrib/mysql_proxy/filters/network/source:config",
     "envoy.filters.network.postgres_proxy":                     "//contrib/postgres_proxy/filters/network/source:config",
     "envoy.filters.network.rocketmq_proxy":                     "//contrib/rocketmq_proxy/filters/network/source:config",

contrib/extensions_metadata.yaml

@@ -13,6 +13,11 @@ envoy.filters.network.kafka_broker:
   - envoy.filters.network
   security_posture: requires_trusted_downstream_and_upstream
   status: wip
+envoy.filters.network.kafka_mesh:
+  categories:
+  - envoy.filters.network
+  security_posture: requires_trusted_downstream_and_upstream
+  status: wip
 envoy.filters.network.rocketmq_proxy:
   categories:
   - envoy.filters.network

contrib/kafka/filters/network/source/mesh/BUILD

@@ -1,5 +1,6 @@
 load(
     "//bazel:envoy_build_system.bzl",
+    "envoy_cc_contrib_extension",
     "envoy_cc_library",
     "envoy_contrib_package",
 )
@@ -10,6 +11,25 @@ licenses(["notice"])  # Apache 2
 envoy_contrib_package()
 
 # Kafka-mesh network filter.
+# Mesh filter public docs: docs/root/configuration/listeners/network_filters/kafka_mesh_filter.rst
+
+envoy_cc_contrib_extension(
+    name = "config_lib",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        "//envoy/registry",
+        "//source/extensions/filters/network/common:factory_base_lib",
+        "@envoy_api//contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha:pkg_cc_proto",
+    ] + select({
+        "//bazel:windows": [],
+        "//conditions:default": [
+            ":filter_lib",
+            ":upstream_config_lib",
+            ":upstream_kafka_facade_lib",
+        ],
+    }),
+)
 
 envoy_cc_library(
     name = "filter_lib",
@@ -121,11 +141,15 @@ envoy_cc_library(
 envoy_cc_library(
     name = "upstream_config_lib",
     srcs = [
+        "upstream_config.cc",
     ],
     hdrs = [
         "upstream_config.h",
     ],
     tags = ["skip_on_windows"],
     deps = [
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+        "@envoy_api//contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha:pkg_cc_proto",
     ],
 )

contrib/kafka/filters/network/source/mesh/config.cc

@@ -0,0 +1,55 @@
+#include "contrib/kafka/filters/network/source/mesh/config.h"
+
+#include "envoy/registry/registry.h"
+#include "envoy/server/filter_config.h"
+#include "envoy/stats/scope.h"
+
+#ifndef WIN32
+#include "contrib/kafka/filters/network/source/mesh/upstream_config.h"
+#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_facade.h"
+#include "contrib/kafka/filters/network/source/mesh/filter.h"
+#else
+#include "envoy/common/exception.h"
+#endif
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+namespace Mesh {
+
+// The mesh filter doesn't do anything special, it just sets up the shared entities.
+// Any extra configuration validation is done in UpstreamKafkaConfiguration constructor.
+Network::FilterFactoryCb KafkaMeshConfigFactory::createFilterFactoryFromProtoTyped(
+    const KafkaMeshProtoConfig& config, Server::Configuration::FactoryContext& context) {
+
+#ifdef WIN32
+  throw EnvoyException("Kafka mesh filter is not supported on Windows");
+#else
+  // Shared configuration (tells us where the upstream clusters are).
+  const UpstreamKafkaConfigurationSharedPtr configuration =
+      std::make_shared<UpstreamKafkaConfigurationImpl>(config);
+
+  // Shared upstream facade (connects us to upstream Kafka clusters).
+  const UpstreamKafkaFacadeSharedPtr upstream_kafka_facade =
+      std::make_shared<UpstreamKafkaFacadeImpl>(*configuration, context.threadLocal(),
+                                                context.api().threadFactory());
+
+  return [configuration, upstream_kafka_facade](Network::FilterManager& filter_manager) -> void {
+    Network::ReadFilterSharedPtr filter =
+        std::make_shared<KafkaMeshFilter>(*configuration, *upstream_kafka_facade);
+    filter_manager.addReadFilter(filter);
+  };
+#endif
+}
+
+/**
+ * Static registration for the Kafka filter. @see RegisterFactory.
+ */
+REGISTER_FACTORY(KafkaMeshConfigFactory, Server::Configuration::NamedNetworkFilterConfigFactory);
+
+} // namespace Mesh
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy

contrib/kafka/filters/network/source/mesh/config.h

@@ -0,0 +1,33 @@
+#pragma once
+
+#include "source/extensions/filters/network/common/factory_base.h"
+
+#include "contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha/kafka_mesh.pb.h"
+#include "contrib/envoy/extensions/filters/network/kafka_mesh/v3alpha/kafka_mesh.pb.validate.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+namespace Mesh {
+
+using KafkaMeshProtoConfig = envoy::extensions::filters::network::kafka_mesh::v3alpha::KafkaMesh;
+
+/**
+ * Config registration for the Kafka mesh filter.
+ */
+class KafkaMeshConfigFactory : public Common::FactoryBase<KafkaMeshProtoConfig> {
+public:
+  KafkaMeshConfigFactory() : FactoryBase("envoy.filters.network.kafka_mesh", true) {}
+
+private:
+  Network::FilterFactoryCb
+  createFilterFactoryFromProtoTyped(const KafkaMeshProtoConfig& config,
+                                    Server::Configuration::FactoryContext& context) override;
+};
+
+} // namespace Mesh
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy

contrib/kafka/filters/network/source/mesh/upstream_config.cc

@@ -0,0 +1,93 @@
+#include "contrib/kafka/filters/network/source/mesh/upstream_config.h"
+
+#include "envoy/common/exception.h"
+
+#include "source/common/common/assert.h"
+
+#include "absl/strings/str_cat.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+namespace Mesh {
+
+using KafkaClusterDefinition =
+    envoy::extensions::filters::network::kafka_mesh::v3alpha::KafkaClusterDefinition;
+using ForwardingRule = envoy::extensions::filters::network::kafka_mesh::v3alpha::ForwardingRule;
+
+UpstreamKafkaConfigurationImpl::UpstreamKafkaConfigurationImpl(const KafkaMeshProtoConfig& config)
+    : advertised_address_{config.advertised_host(), config.advertised_port()} {
+
+  // Processing cluster data.
+  const auto& upstream_clusters = config.upstream_clusters();
+  if (upstream_clusters.empty()) {
+    throw EnvoyException("kafka-mesh filter needs to have at least one upstream Kafka cluster");
+  }
+
+  // Processing cluster configuration.
+  std::map<std::string, ClusterConfig> cluster_name_to_cluster_config;
+  for (const auto& upstream_cluster_definition : upstream_clusters) {
+    const std::string& cluster_name = upstream_cluster_definition.cluster_name();
+
+    // No duplicates are allowed.
+    if (cluster_name_to_cluster_config.find(cluster_name) != cluster_name_to_cluster_config.end()) {
+      throw EnvoyException(
+          absl::StrCat("kafka-mesh filter has multiple Kafka clusters referenced by the same name",
+                       cluster_name));
+    }
+
+    // Upstream client configuration - use all the optional custom configs provided, and then use
+    // the target IPs.
+    std::map<std::string, std::string> producer_configs = {
+        upstream_cluster_definition.producer_config().begin(),
+        upstream_cluster_definition.producer_config().end()};
+    producer_configs["bootstrap.servers"] = upstream_cluster_definition.bootstrap_servers();
+    ClusterConfig cluster_config = {cluster_name, upstream_cluster_definition.partition_count(),
+                                    producer_configs};
+    cluster_name_to_cluster_config[cluster_name] = cluster_config;
+  }
+
+  // Processing forwarding rules.
+  const auto& forwarding_rules = config.forwarding_rules();
+  if (forwarding_rules.empty()) {
+    throw EnvoyException("kafka-mesh filter needs to have at least one forwarding rule");
+  }
+
+  for (const auto& rule : forwarding_rules) {
+    const std::string& target_cluster = rule.target_cluster();
+    ASSERT(rule.trigger_case() == ForwardingRule::TriggerCase::kTopicPrefix);
+    ENVOY_LOG(trace, "Setting up forwarding rule: {} -> {}", rule.topic_prefix(), target_cluster);
+    // Each forwarding rule needs to reference a cluster.
+    if (cluster_name_to_cluster_config.find(target_cluster) ==
+        cluster_name_to_cluster_config.end()) {
+      throw EnvoyException(absl::StrCat(
+          "kafka-mesh filter forwarding rule is referencing unknown upstream Kafka cluster: ",
+          target_cluster));
+    }
+    topic_prefix_to_cluster_config_[rule.topic_prefix()] =
+        cluster_name_to_cluster_config[target_cluster];
+  }
+}
+
+absl::optional<ClusterConfig>
+UpstreamKafkaConfigurationImpl::computeClusterConfigForTopic(const std::string& topic) const {
+  // We find the first matching prefix (this is why ordering is important).
+  for (const auto& it : topic_prefix_to_cluster_config_) {
+    if (topic.rfind(it.first, 0) == 0) {
+      const ClusterConfig cluster_config = it.second;
+      return absl::make_optional(cluster_config);
+    }
+  }
+  return absl::nullopt;
+}
+
+std::pair<std::string, int32_t> UpstreamKafkaConfigurationImpl::getAdvertisedAddress() const {
+  return advertised_address_;
+}
+
+} // namespace Mesh
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy