access logging: introduce critical ALS endpoint

api/envoy/extensions/access_loggers/grpc/v3/BUILD

@@ -6,6 +6,7 @@ licenses(["notice"])  # Apache 2
 
 api_proto_package(
     deps = [
+        "//envoy/config/accesslog/v3:pkg",
         "//envoy/config/core/v3:pkg",
         "@com_github_cncf_udpa//udpa/annotations:pkg",
     ],

api/envoy/extensions/access_loggers/grpc/v3/als.proto

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 package envoy.extensions.access_loggers.grpc.v3;
 
+import "envoy/config/accesslog/v3/accesslog.proto";
 import "envoy/config/core/v3/config_source.proto";
 import "envoy/config/core/v3/grpc_service.proto";
 
@@ -54,7 +55,7 @@ message TcpGrpcAccessLogConfig {
 }
 
 // Common configuration for gRPC access logs.
-// [#next-free-field: 7]
+// [#next-free-field: 10]
 message CommonGrpcAccessLogConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.accesslog.v2.CommonGrpcAccessLogConfig";
@@ -86,4 +87,20 @@ message CommonGrpcAccessLogConfig {
   // <envoy_v3_api_field_data.accesslog.v3.AccessLogCommon.filter_state_objects>`.
   // Logger will call `FilterState::Object::serializeAsProto` to serialize the filter state object.
   repeated string filter_state_objects_to_log = 5;
+
+  // Define the log condition for critical access logs.
+  // Logs that match the filter are not forwarded to the normal endpoint,
+  // but are sent to a special endpoint.
+  config.accesslog.v3.AccessLogFilter critical_buffer_log_filter = 7;
+
+  // The time to wait for an ACK message. If no ACK message is returned after this time,
+  // the message is considered undeliverable and the failed transmission is buffered again.
+  // The re-buffered message will be sent again at the next time a log matching *critical_buffer_log_filter* is queued.
+  google.protobuf.Duration message_ack_timeout = 8
+      [(validate.rules).duration = {gte {nanos: 1000000}}];
+
+  // Size limit (in bytes) of the buffer used to store messages during processing in a
+  // critical logger. A critical logger buffers messages until it receives an ACK from upstream.
+  // The default is 16384.
+  google.protobuf.UInt32Value max_pending_buffer_size_bytes = 9;
 }

api/envoy/service/accesslog/v3/als.proto

@@ -21,12 +21,40 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 service AccessLogService {
   // Envoy will connect and send StreamAccessLogsMessage messages forever. It does not expect any
   // response to be sent as nothing would be done in the case of failure. The server should
-  // disconnect if it expects Envoy to reconnect. In the future we may decide to add a different
-  // API for "critical" access logs in which Envoy will buffer access logs for some period of time
-  // until it gets an ACK so it could then retry. This API is designed for high throughput with the
+  // disconnect if it expects Envoy to reconnect. This API is designed for high throughput with the
   // expectation that it might be lossy.
   rpc StreamAccessLogs(stream StreamAccessLogsMessage) returns (StreamAccessLogsResponse) {
   }
+
+  // This endpoint provides acknowledgment of logs marked as requiring acknowledgment.
+  // The requirement for an acknowledgment can be set in
+  // :ref:`critical_buffer_log_filter <envoy_v3_api_msg_extensions.access_loggers.grpc.v3.CommonGrpcAccessLogConfig.critical_buffer_log_filter>`.
+  // Log messages that match this filter will be guaranteed delivery. In order to guarantee
+  // the arrival, this endpoint performs the following process.
+  //
+  // 1. A response message is returned for each log. The response message includes ACK/NACK status,
+  //    and in case of NACK, the target log is not flushed but buffered by Envoy.
+  // 2. Timeout for response message is set and if no message is returned within a certain time,
+  //    it will be considered as unreachable and buffered by Envoy without flushing the target log. This timeout is set by
+  //    :ref:`message_ack_timeout <envoy_v3_api_msg_extensions.access_loggers.grpc.v3.CommonGrpcAccessLogConfig.message_ack_timeout>`.
+  //
+  // On the ALS receiver side, ACK is expected to be returned to indicate that the log was saved properly,
+  // and NACK is expected to be returned when the log could not be saved due to some error.
+  //
+  // .. attention::
+  //
+  // Buffers for guaranteed reachability can be extremely memory-intensive. Therefore, the following points
+  // should be considered when using this endpoint.
+  //
+  // 1. :ref:`critical_buffer_log_filter <envoy_v3_api_msg_extensions.access_loggers.grpc.v3.CommonGrpcAccessLogConfig.critical_buffer_log_filter>`
+  //    should be set strictly. A loose filter may encourage rapid buffer overwhelm and leading to OOM.
+  // 2. :ref:`max_pending_buffer_size_bytes <envoy_v3_api_msg_extensions.access_loggers.grpc.v3.CommonGrpcAccessLogConfig.max_pending_buffer_size_bytes>`
+  //    should be set appropriately to prevent OOM.
+  // 3. Make sure that ALS receiver is implemented properly. If it is not implemented, all messages will
+  //    be buffered, which may cause OOM soon.
+  rpc CriticalAccessLogs(stream CriticalAccessLogsMessage)
+      returns (stream CriticalAccessLogsResponse) {
+  }
 }
 
 // Empty response for the StreamAccessLogs API. Will never be sent. See below.
@@ -35,6 +63,23 @@ message StreamAccessLogsResponse {
       "envoy.service.accesslog.v2.StreamAccessLogsResponse";
 }
 
+// Response received to identify undelivered or delivered messages in CriticalAccessLogs.
+message CriticalAccessLogsResponse {
+  enum Status {
+    // Indicates that the message has been received.
+    ACK = 0;
+
+    // Indicates that the message has not been received.
+    NACK = 1;
+  }
+
+  // This field is used to indicate the arrival status.
+  Status status = 1;
+
+  // Message ID that identifies a message.
+  uint64 id = 2;
+}
+
 // Stream message for the StreamAccessLogs API. Envoy will open a stream to the server and stream
 // access logs without ever expecting a response.
 message StreamAccessLogsMessage {
@@ -85,3 +130,16 @@ message StreamAccessLogsMessage {
     TCPAccessLogEntries tcp_logs = 3;
   }
 }
+
+// Stream message for the CriticalAccessLogs API.
+// Envoy opens a stream to the server and streams the access log,
+// expecting a response. Each message sent is assigned an individual ID,
+// and the state of the message is tracked based on the ID.
+message CriticalAccessLogsMessage {
+  // The body of the log message sent to CriticalAccessLogs.
+  StreamAccessLogsMessage message = 1;
+
+  // This is an ID to identify the message, and should be added to the Critical Endpoint
+  // response message to uniquely identify the message being ACK/NACKed.
+  uint64 id = 4;
+}

docs/root/version_history/current.rst

@@ -84,6 +84,7 @@ Removed Config or Runtime
 
 New Features
 ------------
+* access log: added :ref:`critical log message <envoy_v3_api_field_service.accesslog.v3.CriticalAccessLogsMessage.message>` to AccessLogService to guarantee log arrival.
 * access_log: added :ref:`METADATA<envoy_v3_api_msg_extensions.formatter.metadata.v3.Metadata>` token to handle all types of metadata (DYNAMIC, CLUSTER, ROUTE).
 * bootstrap: added :ref:`inline_headers <envoy_v3_api_field_config.bootstrap.v3.Bootstrap.inline_headers>` in the bootstrap to make custom inline headers bootstrap configurable.
 * contrib: added new :ref:`contrib images <install_contrib>` which contain contrib extensions.

source/extensions/access_loggers/common/BUILD

@@ -59,5 +59,6 @@ envoy_cc_library(
         "//source/common/protobuf:utility_lib",
         "@com_google_absl//absl/types:optional",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+        "@envoy_api//envoy/service/accesslog/v3:pkg_cc_proto",
     ],
 )

source/extensions/access_loggers/common/grpc_access_logger.h

@@ -5,10 +5,12 @@
 #include "envoy/config/core/v3/config_source.pb.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/grpc/async_client_manager.h"
+#include "envoy/service/accesslog/v3/als.pb.h"
 #include "envoy/singleton/instance.h"
 #include "envoy/stats/scope.h"
 #include "envoy/thread_local/thread_local.h"
 
+#include "source/common/access_log/access_log_impl.h"
 #include "source/common/common/assert.h"
 #include "source/common/grpc/typed_async_client.h"
 #include "source/common/protobuf/utility.h"
@@ -44,14 +46,16 @@ template <typename HttpLogProto, typename TcpLogProto> class GrpcAccessLogger {
   /**
    * Log http access entry.
    * @param entry supplies the access log to send.
+   * @param is_critical determine whether log entry is critical or not.
    */
-  virtual void log(HttpLogProto&& entry) PURE;
+  virtual void log(HttpLogProto&& entry, bool is_critical) PURE;
 
   /**
    * Log tcp access entry.
    * @param entry supplies the access log to send.
+   * @param is_critical determine whether log entry is critical or not.
    */
-  virtual void log(TcpLogProto&& entry) PURE;
+  virtual void log(TcpLogProto&& entry, bool is_critical) PURE;
 };
 
 /**
@@ -146,6 +150,25 @@ struct GrpcAccessLoggerStats {
   ALL_GRPC_ACCESS_LOGGER_STATS(GENERATE_COUNTER_STRUCT)
 };
 
+template <class RequestType> class CriticalAccessLoggerGrpcClient {
+public:
+  virtual ~CriticalAccessLoggerGrpcClient() = default;
+
+  /**
+   * Flush critical messages.
+   */
+  virtual void flush(RequestType message) PURE;
+
+  /**
+   * Whether this client has active stream or not.
+   */
+  virtual bool isStreamStarted() PURE;
+};
+
+template <class RequestType>
+using CriticalAccessLoggerGrpcClientPtr =
+    std::unique_ptr<CriticalAccessLoggerGrpcClient<RequestType>>;
+
 /**
  * Base class for defining a gRPC logger with the `HttpLogProto` and `TcpLogProto` access log
  * entries and `LogRequest` and `LogResponse` gRPC messages.
@@ -165,14 +188,20 @@ class GrpcAccessLogger : public Detail::GrpcAccessLogger<HttpLogProto, TcpLogPro
       : client_(client, service_method), buffer_flush_interval_msec_(buffer_flush_interval_msec),
         flush_timer_(dispatcher.createTimer([this]() {
           flush();
+          flushCriticalMessage();
           flush_timer_->enableTimer(buffer_flush_interval_msec_);
         })),
         max_buffer_size_bytes_(max_buffer_size_bytes),
         stats_({ALL_GRPC_ACCESS_LOGGER_STATS(POOL_COUNTER_PREFIX(scope, access_log_prefix))}) {
     flush_timer_->enableTimer(buffer_flush_interval_msec_);
   }
 
-  void log(HttpLogProto&& entry) {
+  void log(HttpLogProto&& entry, bool is_critical) {
+    if (is_critical) {
+      logCritical(std::move(entry));
+      return;
+    }
+
     if (!canLogMore()) {
       return;
     }
@@ -183,7 +212,7 @@ class GrpcAccessLogger : public Detail::GrpcAccessLogger<HttpLogProto, TcpLogPro
     }
   }
 
-  void log(TcpLogProto&& entry) {
+  void log(TcpLogProto&& entry, bool) {
     approximate_message_size_bytes_ += entry.ByteSizeLong();
     addEntry(std::move(entry));
     if (approximate_message_size_bytes_ >= max_buffer_size_bytes_) {
@@ -201,6 +230,8 @@ class GrpcAccessLogger : public Detail::GrpcAccessLogger<HttpLogProto, TcpLogPro
   virtual void addEntry(HttpLogProto&& entry) PURE;
   virtual void addEntry(TcpLogProto&& entry) PURE;
   virtual void clearMessage() { message_.Clear(); }
+  virtual void flushCriticalMessage() {}
+  virtual void logCritical(HttpLogProto&&) {}
 
   void flush() {
     if (isEmpty()) {