access_loggers: add Fluentd access logger extension (envoyproxy#32320)

* initial fluentd access logger implementation Signed-off-by: ohadvano <[email protected]> * fix Signed-off-by: ohadvano <[email protected]> * fixes Signed-off-by: ohadvano <[email protected]> * fix Signed-off-by: ohadvano <[email protected]> * fix format Signed-off-by: ohadvano <[email protected]> * fix format Signed-off-by: ohadvano <[email protected]> * test Signed-off-by: ohadvano <[email protected]> * format Signed-off-by: ohadvano <[email protected]> * format and test fix Signed-off-by: ohadvano <[email protected]> * revert Signed-off-by: ohadvano <[email protected]> * fix compile_time_options Signed-off-by: ohadvano <[email protected]> * add docs reference Signed-off-by: ohadvano <[email protected]> * use exception safe and add tests Signed-off-by: ohadvano <[email protected]> * fix format Signed-off-by: ohadvano <[email protected]> * const and extension status Signed-off-by: ohadvano <[email protected]> * comment and remove redundant build tags Signed-off-by: ohadvano <[email protected]> * fix format Signed-off-by: ohadvano <[email protected]> * use weak_ptr and pin Signed-off-by: ohadvano <[email protected]> * fix format Signed-off-by: ohadvano <[email protected]> * recrate if expired Signed-off-by: ohadvano <[email protected]> --------- Signed-off-by: ohadvano <[email protected]> Signed-off-by: ohadvano <[email protected]>
ohadvano · Feb 21, 2024 · 949c953 · 949c953
1 parent 35d0c8e
commit 949c953
Show file tree

Hide file tree

Showing 31 changed files with 1,411 additions and 0 deletions.
diff --git a/CODEOWNERS b/CODEOWNERS
@@ -258,6 +258,7 @@ extensions/filters/http/oauth2 @derekargueta @mattklein123
 # proxy protocol
 /*/extensions/filters/listener/proxy_protocol @ggreenway @soulxu
 # access loggers
+/*/extensions/access_loggers/fluentd @ohadvano @wbpcode
 /*/extensions/access_loggers/grpc @wbpcode @cpakulski @giantcroc @gyohuangxin
 # stats
 /*/extensions/stat_sinks/statsd @mattklein123 @suniltheta

diff --git a/api/BUILD b/api/BUILD
@@ -134,6 +134,7 @@ proto_library(
         "//envoy/data/tap/v3:pkg",
         "//envoy/extensions/access_loggers/file/v3:pkg",
         "//envoy/extensions/access_loggers/filters/cel/v3:pkg",
+        "//envoy/extensions/access_loggers/fluentd/v3:pkg",
         "//envoy/extensions/access_loggers/grpc/v3:pkg",
         "//envoy/extensions/access_loggers/open_telemetry/v3:pkg",
         "//envoy/extensions/access_loggers/stream/v3:pkg",

diff --git a/api/envoy/extensions/access_loggers/fluentd/v3/BUILD b/api/envoy/extensions/access_loggers/fluentd/v3/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_xds//udpa/annotations:pkg"],
+)
diff --git a/api/envoy/extensions/access_loggers/fluentd/v3/fluentd.proto b/api/envoy/extensions/access_loggers/fluentd/v3/fluentd.proto
@@ -0,0 +1,70 @@
+syntax = "proto3";
+
+package envoy.extensions.access_loggers.fluentd.v3;
+
+import "google/protobuf/duration.proto";
+import "google/protobuf/struct.proto";
+import "google/protobuf/wrappers.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.access_loggers.fluentd.v3";
+option java_outer_classname = "FluentdProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/access_loggers/fluentd/v3;fluentdv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Fluentd access log]
+
+// Configuration for the *envoy.access_loggers.fluentd* :ref:`AccessLog <envoy_v3_api_msg_config.accesslog.v3.AccessLog>`.
+// This access log extension will send the emitted access logs over a TCP connection to an upstream that is accepting
+// 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]
+message FluentdAccessLogConfig {
+  // The upstream cluster to connect to for streaming the Fluentd messages.
+  string cluster = 1 [(validate.rules).string = {min_len: 1}];
+
+  // A tag is a string separated with '.' (e.g. log.type) to categorize events.
+  // See: https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#message-modes
+  string tag = 2 [(validate.rules).string = {min_len: 1}];
+
+  // The prefix to use when emitting :ref:`statistics <config_access_log_stats>`.
+  string stat_prefix = 3 [(validate.rules).string = {min_len: 1}];
+
+  // Interval for flushing access logs to the TCP stream. Logger will flush requests every time
+  // this interval is elapsed, or when batch size limit is hit, whichever comes first. Defaults to
+  // 1 second.
+  google.protobuf.Duration buffer_flush_interval = 4 [(validate.rules).duration = {gt {}}];
+
+  // Soft size limit in bytes for access log entries buffer. The logger will buffer requests until
+  // this limit it hit, or every time flush interval is elapsed, whichever comes first. When the buffer
+  // limit is hit, the logger will immediately flush the buffer contents. Setting it to zero effectively
+  // disables the batching. Defaults to 16384.
+  google.protobuf.UInt32Value buffer_size_bytes = 5;
+
+  // A struct that represents the record that is sent for each log entry.
+  // https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#entry
+  // Values are rendered as strings, numbers, or boolean values as appropriate.
+  // Nested JSON objects may be produced by some command operators (e.g. FILTER_STATE or DYNAMIC_METADATA).
+  // See :ref:`format string<config_access_log_format_strings>` documentation for a specific command operator details.
+  //
+  // .. validated-code-block:: yaml
+  //   :type-name: envoy.extensions.access_loggers.fluentd.v3.FluentdAccessLogConfig
+  //
+  //   record:
+  //     status: "%RESPONSE_CODE%"
+  //     message: "%LOCAL_REPLY_BODY%"
+  //
+  // The following msgpack record would be created:
+  //
+  // .. code-block:: json
+  //
+  //  {
+  //    "status": 500,
+  //    "message": "My error message"
+  //  }
+  google.protobuf.Struct record = 6 [(validate.rules).message = {required: true}];
+}
diff --git a/api/versioning/BUILD b/api/versioning/BUILD
@@ -72,6 +72,7 @@ proto_library(
         "//envoy/data/tap/v3:pkg",
         "//envoy/extensions/access_loggers/file/v3:pkg",
         "//envoy/extensions/access_loggers/filters/cel/v3:pkg",
+        "//envoy/extensions/access_loggers/fluentd/v3:pkg",
         "//envoy/extensions/access_loggers/grpc/v3:pkg",
         "//envoy/extensions/access_loggers/open_telemetry/v3:pkg",
         "//envoy/extensions/access_loggers/stream/v3:pkg",

diff --git a/bazel/external/msgpack.BUILD b/bazel/external/msgpack.BUILD
@@ -0,0 +1,16 @@
+licenses(["notice"])  # Apache 2
+
+cc_library(
+    name = "msgpack",
+    srcs = glob([
+        "src/*.c",
+        "include/**/*.h",
+        "include/**/*.hpp",
+    ]),
+    defines = ["MSGPACK_NO_BOOST"],
+    includes = [
+        "include",
+    ],
+    strip_include_prefix = "include",
+    visibility = ["//visibility:public"],
+)
diff --git a/bazel/repositories.bzl b/bazel/repositories.bzl
@@ -318,6 +318,7 @@ def envoy_dependencies(skip_targets = []):
     _com_github_libevent_libevent()
     _com_github_luajit_luajit()
     _com_github_nghttp2_nghttp2()
+    _com_github_msgpack_cpp()
     _com_github_skyapm_cpp2sky()
     _com_github_nodejs_http_parser()
     _com_github_alibaba_hessian2_codec()
@@ -735,6 +736,16 @@ def _com_github_nghttp2_nghttp2():
         actual = "@envoy//bazel/foreign_cc:nghttp2",
     )
 
+def _com_github_msgpack_cpp():
+    external_http_archive(
+        name = "com_github_msgpack_cpp",
+        build_file = "@envoy//bazel/external:msgpack.BUILD",
+    )
+    native.bind(
+        name = "msgpack",
+        actual = "@com_github_msgpack_cpp//:msgpack",
+    )
+
 def _io_hyperscan():
     external_http_archive(
         name = "io_hyperscan",

diff --git a/bazel/repository_locations.bzl b/bazel/repository_locations.bzl
@@ -778,6 +778,21 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         license = "MIT",
         license_url = "https://github.com/jbeder/yaml-cpp/blob/{version}/LICENSE",
     ),
+    com_github_msgpack_cpp = dict(
+        project_name = "msgpack for C/C++",
+        project_desc = "MessagePack is an efficient binary serialization format",
+        project_url = "https://github.com/msgpack/msgpack-c",
+        version = "6.1.0",
+        sha256 = "23ede7e93c8efee343ad8c6514c28f3708207e5106af3b3e4969b3a9ed7039e7",
+        strip_prefix = "msgpack-cxx-{version}",
+        urls = ["https://github.com/msgpack/msgpack-c/releases/download/cpp-{version}/msgpack-cxx-{version}.tar.gz"],
+        use_category = ["observability_ext"],
+        extensions = ["envoy.access_loggers.fluentd"],
+        release_date = "2023-07-08",
+        cpe = "cpe:2.3:a:messagepack:messagepack:*",
+        license = "Boost",
+        license_url = "https://github.com/msgpack/msgpack-c/blob/cpp-{version}/LICENSE_1_0.txt",
+    ),
     com_github_google_jwt_verify = dict(
         project_name = "jwt_verify_lib",
         project_desc = "JWT verification library for C++",

diff --git a/changelogs/current.yaml b/changelogs/current.yaml
@@ -204,6 +204,10 @@ new_features:
     Envoy will select the host with the fewest active requests from the entire host set rather than
     :ref:`choice_count <envoy_v3_api_msg_extensions.load_balancing_policies.least_request.v3.LeastRequest>`
     random choices.
+- area: access_loggers
+  change: |
+    Added :ref:`Fluentd access logger <envoy_v3_api_msg_extensions.access_loggers.fluentd.v3.FluentdAccessLogConfig>`
+    to support flushing access logs in `Fluentd format <https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1>`_.
 - area: redis
   change: |
     Added support for the ``ECHO`` command.

diff --git a/docs/root/configuration/observability/access_log/stats.rst b/docs/root/configuration/observability/access_log/stats.rst
@@ -33,3 +33,17 @@ The file access log has statistics rooted at the *filesystem.* namespace.
   flushed_by_timer, Counter, Total number of times internal flush buffers are written to a file due to flush timeout
   reopen_failed, Counter, Total number of times a file was failed to be opened
   write_total_buffered, Gauge, Current total size of internal flush buffer in bytes
+
+Fluentd access log statistics
+-----------------------------
+
+The Fluentd access log has statistics rooted at the *access_logs.fluentd.<stat_prefix>.* namespace.
+
+.. csv-table::
+  :header: Name, Type, Description
+  :widths: 1, 1, 2
+
+  entries_lost, Counter, Total number of times an access log entry was discarded due to unavailable connection.
+  entries_buffered, Counter, Total number of entries (access log record) that was buffered/
+  events_sent, Counter, Total number of events (Fluentd Forward Mode events) sent to the upstream.
+  connections_closed, Counter, Total number of times a connection to the upstream cluster was closed.
diff --git a/docs/root/intro/arch_overview/observability/access_logging.rst b/docs/root/intro/arch_overview/observability/access_logging.rst
@@ -138,6 +138,15 @@ Stderr
   response headers.
 * Writes to the standard error of the process. It works in all platforms.
 
+Fluentd
+********
+
+* Flush access logs over a TCP connection to an upstream that is accepting the Fluentd Forward Protocol as described in:
+  `Fluentd Forward Protocol Specification <https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1>`_.
+* The data sent over the wire is a stream of
+  `Fluentd Forward Mode events <https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1#forward-mode>`_
+  which may contain one or more access log entries (depending on the flushing interval and other configuration parameters).
+
 Further reading
 ---------------
 
@@ -148,3 +157,4 @@ Further reading
 * OpenTelemetry (gRPC) :ref:`LogsService <envoy_v3_api_msg_extensions.access_loggers.open_telemetry.v3.OpenTelemetryAccessLogConfig>`
 * Stdout :ref:`access log sink <envoy_v3_api_msg_extensions.access_loggers.stream.v3.StdoutAccessLog>`
 * Stderr :ref:`access log sink <envoy_v3_api_msg_extensions.access_loggers.stream.v3.StderrAccessLog>`
+* Fluentd :ref:`access log sink <envoy_v3_api_msg_extensions.access_loggers.fluentd.v3.FluentdAccessLogConfig>`
diff --git a/source/common/json/json_internal.cc b/source/common/json/json_internal.cc
@@ -788,6 +788,10 @@ std::string Factory::serialize(absl::string_view str) {
   return j.dump();
 }
 
+std::vector<uint8_t> Factory::jsonToMsgpack(const std::string& json_string) {
+  return nlohmann::json::to_msgpack(nlohmann::json::parse(json_string, nullptr, false));
+}
+
 } // namespace Nlohmann
 } // namespace Json
 } // namespace Envoy
diff --git a/source/common/json/json_internal.h b/source/common/json/json_internal.h
@@ -38,6 +38,13 @@ class Factory {
    * @return A string suitable for inclusion in a JSON stream, including double-quotes.
    */
   static std::string serialize(absl::string_view str);
+
+  /*
+   * Serializes a JSON string to a byte vector using the MessagePack serialization format.
+   * If the provided JSON string is invalid, an empty vector will be returned.
+   * See: https://github.com/msgpack/msgpack/blob/master/spec.md
+   */
+  static std::vector<uint8_t> jsonToMsgpack(const std::string& json);
 };
 
 } // namespace Nlohmann

diff --git a/source/common/json/json_loader.cc b/source/common/json/json_loader.cc
@@ -18,5 +18,9 @@ ObjectSharedPtr Factory::loadFromProtobufStruct(const ProtobufWkt::Struct& proto
   return Nlohmann::Factory::loadFromProtobufStruct(protobuf_struct);
 }
 
+std::vector<uint8_t> Factory::jsonToMsgpack(const std::string& json) {
+  return Nlohmann::Factory::jsonToMsgpack(json);
+}
+
 } // namespace Json
 } // namespace Envoy
diff --git a/source/common/json/json_loader.h b/source/common/json/json_loader.h
@@ -27,6 +27,13 @@ class Factory {
    * Constructs a Json Object from a Protobuf struct.
    */
   static ObjectSharedPtr loadFromProtobufStruct(const ProtobufWkt::Struct& protobuf_struct);
+
+  /*
+   * Serializes a JSON string to a byte vector using the MessagePack serialization format.
+   * If the provided JSON string is invalid, an empty vector will be returned.
+   * See: https://github.com/msgpack/msgpack/blob/master/spec.md
+   */
+  static std::vector<uint8_t> jsonToMsgpack(const std::string& json);
 };
 
 } // namespace Json

diff --git a/source/extensions/access_loggers/fluentd/BUILD b/source/extensions/access_loggers/fluentd/BUILD
@@ -0,0 +1,51 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "substitution_formatter_lib",
+    srcs = ["substitution_formatter.cc"],
+    hdrs = ["substitution_formatter.h"],
+    deps = [
+        "//envoy/formatter:substitution_formatter_interface",
+        "//source/common/json:json_loader_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "fluentd_access_log_lib",
+    srcs = ["fluentd_access_log_impl.cc"],
+    hdrs = ["fluentd_access_log_impl.h"],
+    external_deps = [
+        "msgpack",
+    ],
+    deps = [
+        ":substitution_formatter_lib",
+        "//envoy/access_log:access_log_interface",
+        "//source/common/access_log:access_log_lib",
+        "//source/extensions/access_loggers/common:access_log_base",
+        "@envoy_api//envoy/extensions/access_loggers/fluentd/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":fluentd_access_log_lib",
+        ":substitution_formatter_lib",
+        "//envoy/access_log:access_log_config_interface",
+        "//envoy/registry",
+        "//source/common/config:config_provider_lib",
+        "//source/common/formatter:substitution_format_string_lib",
+        "//source/common/protobuf",
+    ],
+)