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: 9]
 message CommonGrpcAccessLogConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.accesslog.v2.CommonGrpcAccessLogConfig";
@@ -86,4 +87,23 @@ 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 as a filter to ensure that the destination is reached.
+  // Logs that match the filter are not forwarded to the normal endpoint,
+  // but are sent to a special endpoint. This endpoint will return ACK/NACK as a response,
+  // and Envoy will perform buffering upon receiving a NACK. The buffered message is
+  // resend until it is reached. Failure to send here means that
+  //
+  // 1. when a NACK message corresponding to the message sent is received
+  // 2. when the ACK message corresponding to the sent message buffer could not be
+  //    received until the message_ack_timeout is reached.
+  //
+  // It is recommended to set a strict filter because setting a loose filter here may cause
+  // extreme pressure on the Envoy buffer in cases where the log aggregation cluster is repeatedly restarted.
+  config.accesslog.v3.AccessLogFilter buffer_log_filter = 7
+      [(validate.rules).enum = {defined_only: true}];
+
+  // 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.
+  google.protobuf.Duration message_ack_timeout = 8 [(validate.rules).duration = {gt {}}];
 }

api/envoy/extensions/access_loggers/grpc/v4alpha/BUILD

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

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

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 package envoy.extensions.access_loggers.grpc.v4alpha;
 
+import "envoy/config/accesslog/v4alpha/accesslog.proto";
 import "envoy/config/core/v4alpha/config_source.proto";
 import "envoy/config/core/v4alpha/grpc_service.proto";
 
@@ -54,7 +55,7 @@ message TcpGrpcAccessLogConfig {
 }
 
 // Common configuration for gRPC access logs.
-// [#next-free-field: 7]
+// [#next-free-field: 9]
 message CommonGrpcAccessLogConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.extensions.access_loggers.grpc.v3.CommonGrpcAccessLogConfig";
@@ -86,4 +87,23 @@ 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 as a filter to ensure that the destination is reached.
+  // Logs that match the filter are not forwarded to the normal endpoint,
+  // but are sent to a special endpoint. This endpoint will return ACK/NACK as a response,
+  // and Envoy will perform buffering upon receiving a NACK. The buffered message is
+  // resend until it is reached. Failure to send here means that
+  //
+  // 1. when a NACK message corresponding to the message sent is received
+  // 2. when the ACK message corresponding to the sent message buffer could not be
+  //    received until the message_ack_timeout is reached.
+  //
+  // It is recommended to set a strict filter because setting a loose filter here may cause
+  // extreme pressure on the Envoy buffer in cases where the log aggregation cluster is repeatedly restarted.
+  config.accesslog.v4alpha.AccessLogFilter buffer_log_filter = 7
+      [(validate.rules).enum = {defined_only: true}];
+
+  // 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.
+  google.protobuf.Duration message_ack_timeout = 8 [(validate.rules).duration = {gt {}}];
 }

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

@@ -27,6 +27,10 @@ service AccessLogService {
   // expectation that it might be lossy.
   rpc StreamAccessLogs(stream StreamAccessLogsMessage) returns (StreamAccessLogsResponse) {
   }
+
+  rpc BufferedCriticalAccessLogs(StreamAccessLogsMessage)
+      returns (BufferedCriticalAccessLogsResponse) {
+  }
 }
 
 // Empty response for the StreamAccessLogs API. Will never be sent. See below.
@@ -35,6 +39,20 @@ message StreamAccessLogsResponse {
       "envoy.service.accesslog.v2.StreamAccessLogsResponse";
 }
 
+// Response received to identify undelivered or delivered messages in BufferedCriticalAccessLogs.
+message BufferedCriticalAccessLogsResponse {
+  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;
+}
+
 // 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 {

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

@@ -27,6 +27,10 @@ service AccessLogService {
   // expectation that it might be lossy.
   rpc StreamAccessLogs(stream StreamAccessLogsMessage) returns (StreamAccessLogsResponse) {
   }
+
+  rpc BufferedCriticalAccessLogs(StreamAccessLogsMessage)
+      returns (BufferedCriticalAccessLogsResponse) {
+  }
 }
 
 // Empty response for the StreamAccessLogs API. Will never be sent. See below.
@@ -35,6 +39,23 @@ message StreamAccessLogsResponse {
       "envoy.service.accesslog.v3.StreamAccessLogsResponse";
 }
 
+// Response received to identify undelivered or delivered messages in BufferedCriticalAccessLogs.
+message BufferedCriticalAccessLogsResponse {
+  option (udpa.annotations.versioning).previous_message_type =
+      "envoy.service.accesslog.v3.BufferedCriticalAccessLogsResponse";
+
+  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;
+}
+
 // 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 {

source/extensions/access_loggers/common/BUILD

@@ -59,5 +59,7 @@ 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/extensions/access_loggers/grpc/v3:pkg_cc_proto",
+        "@envoy_api//envoy/service/accesslog/v3:pkg_cc_proto",
     ],
 )