Kafka codec

api/bazel/repositories.bzl

@@ -31,6 +31,11 @@ def api_dependencies():
         locations = REPOSITORY_LOCATIONS,
         build_file_content = OPENCENSUSTRACE_BUILD_CONTENT,
     )
+    envoy_http_archive(
+        name = "kafka_source",
+        locations = REPOSITORY_LOCATIONS,
+        build_file_content = KAFKASOURCE_BUILD_CONTENT,
+    )
 
 GOOGLEAPIS_BUILD_CONTENT = """
 load("@com_google_protobuf//:protobuf.bzl", "cc_proto_library", "py_proto_library")
@@ -285,3 +290,15 @@ go_proto_library(
     visibility = ["//visibility:public"],
 )
 """
+
+KAFKASOURCE_BUILD_CONTENT = """
+
+filegroup(
+    name = "request_protocol_files",
+    srcs = glob([
+        "*Request.json",
+    ]),
+    visibility = ["//visibility:public"],
+)
+
+"""

api/bazel/repository_locations.bzl

@@ -16,6 +16,8 @@ GOOGLEAPIS_SHA = "16f5b2e8bf1e747a32f9a62e211f8f33c94645492e9bbd72458061d9a9de1f
 PROMETHEUS_GIT_SHA = "99fa1f4be8e564e8a6b613da7fa6f46c9edafc6c"  # Nov 17, 2017
 PROMETHEUS_SHA = "783bdaf8ee0464b35ec0c8704871e1e72afa0005c3f3587f65d9d6694bf3911b"
 
+KAFKA_SOURCE_SHA = "ae7a1696c0a0302b43c5b21e515c37e6ecd365941f68a510a7e442eebddf39a1"  # 2.2.0-rc2
+
 REPOSITORY_LOCATIONS = dict(
     bazel_skylib = dict(
         sha256 = BAZEL_SKYLIB_SHA256,
@@ -48,4 +50,9 @@ REPOSITORY_LOCATIONS = dict(
         strip_prefix = "opencensus-proto-" + OPENCENSUS_RELEASE + "/src/opencensus/proto/trace/v1",
         urls = ["https://github.com/census-instrumentation/opencensus-proto/archive/v" + OPENCENSUS_RELEASE + ".tar.gz"],
     ),
+    kafka_source = dict(
+        sha256 = KAFKA_SOURCE_SHA,
+        strip_prefix = "kafka-2.2.0-rc2/clients/src/main/resources/common/message",
+        urls = ["https://github.com/apache/kafka/archive/2.2.0-rc2.zip"],
+    ),
 )

source/common/common/logger.h

@@ -37,6 +37,7 @@ namespace Logger {
   FUNCTION(http2)                \
   FUNCTION(hystrix)              \
   FUNCTION(init)                 \
+  FUNCTION(kafka)                \
   FUNCTION(lua)                  \
   FUNCTION(main)                 \
   FUNCTION(misc)                 \

source/extensions/extensions_build_config.bzl

@@ -69,6 +69,9 @@ EXTENSIONS = {
     "envoy.filters.network.echo":                       "//source/extensions/filters/network/echo:config",
     "envoy.filters.network.ext_authz":                  "//source/extensions/filters/network/ext_authz:config",
     "envoy.filters.network.http_connection_manager":    "//source/extensions/filters/network/http_connection_manager:config",
+    # NOTE: Kafka filter does not have a proper filter implemented right now. We are referencing to
+    #       codec implementation that is going to be used by the filter.
+    "envoy.filters.network.kafka":                      "//source/extensions/filters/network/kafka:kafka_request_codec_lib",
     "envoy.filters.network.mongo_proxy":                "//source/extensions/filters/network/mongo_proxy:config",
     "envoy.filters.network.mysql_proxy":                "//source/extensions/filters/network/mysql_proxy:config",
     "envoy.filters.network.ratelimit":                  "//source/extensions/filters/network/ratelimit:config",

source/extensions/filters/network/kafka/BUILD

@@ -0,0 +1,137 @@
+licenses(["notice"])  # Apache 2
+
+# Kafka network filter.
+# Public docs: docs/root/configuration/network_filters/kafka_filter.rst
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "kafka_request_codec_lib",
+    srcs = ["request_codec.cc"],
+    hdrs = [
+        "codec.h",
+        "request_codec.h",
+    ],
+    deps = [
+        ":kafka_request_parser_lib",
+        "//source/common/buffer:buffer_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_request_parser_lib",
+    srcs = [
+        "external/kafka_request_resolver.cc",
+        "kafka_request_parser.cc",
+    ],
+    hdrs = [
+        "external/requests.h",
+        "kafka_request_parser.h",
+    ],
+    deps = [
+        ":kafka_request_lib",
+        ":parser_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "kafka_request_lib",
+    srcs = [
+    ],
+    hdrs = [
+        "kafka_request.h",
+    ],
+    deps = [
+        ":serialization_lib",
+    ],
+)
+
+genrule(
+    name = "kafka_generated_source",
+    srcs = [
+        "@kafka_source//:request_protocol_files",
+    ],
+    outs = [
+        "external/requests.h",
+        "external/kafka_request_resolver.cc",
+    ],
+    cmd = """
+      ./$(location :kafka_code_generator) generate-source \
+        $(location external/requests.h) $(location external/kafka_request_resolver.cc) \
+        $(SRCS)
+    """,
+    tools = [
+        ":kafka_code_generator",
+    ],
+)
+
+py_binary(
+    name = "kafka_code_generator",
+    srcs = ["protocol_code_generator/kafka_generator.py"],
+    data = glob(["protocol_code_generator/*.j2"]),
+    main = "protocol_code_generator/kafka_generator.py",
+    deps = ["@com_github_pallets_jinja//:jinja2"],
+)
+
+envoy_cc_library(
+    name = "parser_lib",
+    hdrs = ["parser.h"],
+    deps = [
+        "//source/common/common:minimal_logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "serialization_lib",
+    hdrs = [
+        "external/serialization_composite.h",
+        "serialization.h",
+    ],
+    deps = [
+        ":kafka_types_lib",
+        "//include/envoy/buffer:buffer_interface",
+        "//source/common/common:byte_order_lib",
+    ],
+)
+
+genrule(
+    name = "serialization_composite_generated_source",
+    srcs = [],
+    outs = [
+        "external/serialization_composite.h",
+    ],
+    cmd = """
+      ./$(location :serialization_composite_generator) generate-source \
+      $(location external/serialization_composite.h)
+    """,
+    tools = [
+        ":serialization_composite_generator",
+    ],
+)
+
+py_binary(
+    name = "serialization_composite_generator",
+    srcs = ["serialization_code_generator/serialization_composite_generator.py"],
+    data = glob(["serialization_code_generator/*.j2"]),
+    main = "serialization_code_generator/serialization_composite_generator.py",
+    deps = ["@com_github_pallets_jinja//:jinja2"],
+)
+
+envoy_cc_library(
+    name = "kafka_types_lib",
+    hdrs = [
+        "kafka_types.h",
+    ],
+    external_deps = ["abseil_optional"],
+    deps = [
+        "//source/common/common:macros",
+    ],
+)

source/extensions/filters/network/kafka/codec.h

@@ -0,0 +1,43 @@
+#pragma once
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+/**
+ * Kafka message decoder.
+ */
+class MessageDecoder {
+public:
+  virtual ~MessageDecoder() = default;
+
+  /**
+   * Processes given buffer attempting to decode messages contained within.
+   * @param data buffer instance.
+   */
+  virtual void onData(Buffer::Instance& data) PURE;
+};
+
+/**
+ * Kafka message encoder.
+ * @param MessageType encoded message type (request or response).
+ */
+template <typename MessageType> class MessageEncoder {
+public:
+  virtual ~MessageEncoder() = default;
+
+  /**
+   * Encodes given message.
+   * @param message message to be encoded.
+   */
+  virtual void encode(const MessageType& message) PURE;
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/network/kafka/kafka_request.h

@@ -0,0 +1,112 @@
+#pragma once
+
+#include "envoy/common/exception.h"
+
+#include "extensions/filters/network/kafka/external/serialization_composite.h"
+#include "extensions/filters/network/kafka/serialization.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace Kafka {
+
+/**
+ * Represents fields that are present in every Kafka request message.
+ * @see http://kafka.apache.org/protocol.html#protocol_messages
+ */
+struct RequestHeader {
+  int16_t api_key_;
+  int16_t api_version_;
+  int32_t correlation_id_;
+  NullableString client_id_;
+
+  bool operator==(const RequestHeader& rhs) const {
+    return api_key_ == rhs.api_key_ && api_version_ == rhs.api_version_ &&
+           correlation_id_ == rhs.correlation_id_ && client_id_ == rhs.client_id_;
+  };
+};
+
+/**
+ * Carries information that could be extracted during the failed parse.
+ */
+class RequestParseFailure {
+public:
+  RequestParseFailure(const RequestHeader& request_header) : request_header_{request_header} {};
+
+  /**
+   * Request's header.
+   */
+  const RequestHeader request_header_;
+};
+
+typedef std::shared_ptr<RequestParseFailure> RequestParseFailureSharedPtr;
+
+/**
+ * Abstract Kafka request.
+ * Contains data present in every request (the header with request key, version, etc.).
+ * @see http://kafka.apache.org/protocol.html#protocol_messages
+ */
+class AbstractRequest {
+public:
+  virtual ~AbstractRequest() = default;
+
+  /**
+   * Constructs a request with given header data.
+   * @param request_header request's header.
+   */
+  AbstractRequest(const RequestHeader& request_header) : request_header_{request_header} {};
+
+  /**
+   * Encode the contents of this message into a given buffer.
+   * @param dst buffer instance to keep serialized message
+   */
+  virtual size_t encode(Buffer::Instance& dst) const PURE;
+
+  /**
+   * Request's header.
+   */
+  const RequestHeader request_header_;
+};
+
+typedef std::shared_ptr<AbstractRequest> AbstractRequestSharedPtr;
+
+/**
+ * Concrete request that carries data particular to given request type.
+ * @param Data concrete request data type.
+ */
+template <typename Data> class Request : public AbstractRequest {
+public:
+  /**
+   * Request header fields need to be initialized by user in case of newly created requests.
+   */
+  Request(const RequestHeader& request_header, const Data& data)
+      : AbstractRequest{request_header}, data_{data} {};
+
+  /**
+   * Encodes given request into a buffer, with any extra configuration carried by the context.
+   */
+  size_t encode(Buffer::Instance& dst) const override {
+    EncodingContext context{request_header_.api_version_};
+    size_t written{0};
+    // Encode request header.
+    written += context.encode(request_header_.api_key_, dst);
+    written += context.encode(request_header_.api_version_, dst);
+    written += context.encode(request_header_.correlation_id_, dst);
+    written += context.encode(request_header_.client_id_, dst);
+    // Encode request-specific data.
+    written += context.encode(data_, dst);
+    return written;
+  }
+
+  bool operator==(const Request<Data>& rhs) const {
+    return request_header_ == rhs.request_header_ && data_ == rhs.data_;
+  };
+
+private:
+  const Data data_;
+};
+
+} // namespace Kafka
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy