Support Gauge in Stats AccessLogger

api/envoy/extensions/access_loggers/stats/v3/BUILD

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

api/envoy/extensions/access_loggers/stats/v3/stats.proto

@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package envoy.extensions.access_loggers.stats.v3;
 
+import "envoy/data/accesslog/v3/accesslog.proto";
+
 import "google/protobuf/wrappers.proto";
 
 import "xds/annotations/v3/status.proto";
@@ -27,6 +29,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 //   leading to a denial of service in Envoy, or can overwhelm any configured
 //   stat sinks by sending too many unique metrics.
 
+// [#next-free-field: 6]
 message Config {
   option (xds.annotations.v3.message_status).work_in_progress = true;
 
@@ -91,6 +94,58 @@ message Config {
     google.protobuf.UInt64Value value_fixed = 3 [(validate.rules).uint64 = {gt: 0}];
   }
 
+  // Configuration for a gauge stat. Gauges can be used to add, subtract, or set
+  // values, and are useful for tracking concurrency or other mutable values
+  // over time.
+  // [#next-free-field: 6]
+  message Gauge {
+    // The Set operation config.
+    message Set {
+      // The access log type to trigger the operation.
+      data.accesslog.v3.AccessLogType log_type = 1 [(validate.rules).enum = {defined_only: true}];
+    }
+
+    // The PairedAddSubtract operation config.
+    // Usage restrictions:
+    //
+    // 1. We only support add first then subtract logic and we rely on the symmetrical log types
+    //    (e.g., DownstreamStart/DownstreamEnd) to increment and decrement the gauge.
+    // 2. During runtime, sub_log_type will execute if and only if add_log_type operation has
+    //    been done, tracked by inflight counter in filter state.
+    // 3. If the add_log_type operation was executed, the sub_log_type will happen when the
+    //    stream/connection is closed, even if the configured log type didn't happen.
+    message PairedAddSubtract {
+      // The access log type to trigger the add operation.
+      data.accesslog.v3.AccessLogType add_log_type = 1
+          [(validate.rules).enum = {defined_only: true}];
+
+      // The access log type to trigger the subtract operation.
+      data.accesslog.v3.AccessLogType sub_log_type = 2
+          [(validate.rules).enum = {defined_only: true}];
+    }
+
+    // The name and tags of this gauge.
+    Stat stat = 1 [(validate.rules).message = {required: true}];
+
+    // The format string for the value of this gauge, using :ref:`command
+    // operators <config_access_log_command_operators>`. This must evaluate to a
+    // positive number.
+    string value_format = 2
+        [(validate.rules).string = {prefix: "%" suffix: "%" ignore_empty: true}];
+
+    // A fixed value to add/subtract/set to this gauge.
+    // One of ``value_format`` or ``value_fixed`` must be configured.
+    google.protobuf.UInt64Value value_fixed = 3 [(validate.rules).uint64 = {gt: 0}];
+
+    // The PairedAddSubtract operation.
+    // Only one of PairedAddSubtract and Set can be defined.
+    PairedAddSubtract add_subtract = 4;
+
+    // The Set operation.
+    // Only one of PairedAddSubtract and Set can be defined.
+    Set set = 5;
+  }
+
   // The stat prefix for the generated stats.
   string stat_prefix = 1 [(validate.rules).string = {min_len: 1}];
 
@@ -99,4 +154,7 @@ message Config {
 
   // The counters this logger will emit.
   repeated Counter counters = 4;
+
+  // The gauges this logger will emit.
+  repeated Gauge gauges = 5;
 }

changelogs/current.yaml

@@ -141,6 +141,9 @@ new_features:
 - area: mcp_router
   change: |
     Added support for MCP logging method ``logging/setLevel``.
+- area: access_log
+  change: |
+    Supported gauge in the :ref:`stats access logger <envoy_v3_api_msg_extensions.access_loggers.stats.v3.Config>`.
 - area: network
   change: |
     Added support access logging in network filters like http filters, by allowing network filters to register as access logger

source/extensions/access_loggers/stats/BUILD

@@ -19,6 +19,7 @@ envoy_cc_library(
         "//source/common/access_log:access_log_lib",
         "//source/common/formatter:substitution_format_string_lib",
         "//source/extensions/access_loggers/common:access_log_base",
+        "@envoy_api//envoy/data/accesslog/v3:pkg_cc_proto",
         "@envoy_api//envoy/extensions/access_loggers/stats/v3:pkg_cc_proto",
     ],
 )

source/extensions/access_loggers/stats/stats.cc

@@ -1,5 +1,9 @@
 #include "source/extensions/access_loggers/stats/stats.h"
 
+#include "envoy/data/accesslog/v3/accesslog.pb.h"
+#include "envoy/stats/scope.h"
+#include "envoy/stream_info/filter_state.h"
+
 #include "source/common/formatter/substitution_formatter.h"
 
 namespace Envoy {
@@ -8,6 +12,69 @@ namespace AccessLoggers {
 namespace StatsAccessLog {
 
 namespace {
+
+class AccessLogState : public StreamInfo::FilterState::Object {
+public:
+  AccessLogState(Stats::ScopeSharedPtr scope) : scope_(std::move(scope)) {}
+
+  // When the request is destroyed, we need to subtract the value from the gauge.
+  // We need to look up the gauge again in the scope because it might have been evicted.
+  // The gauge object itself is kept alive by the shared_ptr in the state, so we can access its
+  // name and tags to re-lookup/re-create it in the scope.
+  ~AccessLogState() override {
+    for (const auto& [gauge_ptr, state] : inflight_gauges_) {
+      // TODO(taoxuy):  make this as an accessor of the
+      // Stat class.
+      Stats::StatNameTagVector tag_names;
+      state.gauge_->iterateTagStatNames(
+          [&tag_names](Stats::StatName name, Stats::StatName value) -> bool {
+            tag_names.emplace_back(name, value);
+            return true;
+          });
+
+      // Using state.gauge_->statName() directly would be incorrect because it returns the fully
+      // qualified name (including tags). Passing this full name to scope_->gaugeFromStatName(...)
+      // would cause the scope to attempt tag extraction on the full name. Since the tags in
+      // AccessLogState are often dynamic and not configured in the global tag extractors, this
+      // extraction would likely fail to identify the tags correctly, resulting in a gauge with a
+      // different identity (the full name as the stat name and no tags).
+      auto& gauge = scope_->gaugeFromStatNameWithTags(
+          state.gauge_->tagExtractedStatName(), tag_names, Stats::Gauge::ImportMode::Accumulate);
+      gauge.sub(state.value_);
+    }
+  }
+
+  void addInflightGauge(Stats::Gauge* gauge, uint64_t value) {
+    inflight_gauges_.try_emplace(gauge, Stats::GaugeSharedPtr(gauge), value);
+  }
+
+  absl::optional<uint64_t> removeInflightGauge(Stats::Gauge* gauge) {
+    auto it = inflight_gauges_.find(gauge);
+    if (it == inflight_gauges_.end()) {
+      return absl::nullopt;
+    }
+    uint64_t value = it->second.value_;
+    inflight_gauges_.erase(it);
+    return value;
+  }
+
+  static constexpr absl::string_view key() { return "envoy.access_loggers.stats.access_log_state"; }
+
+private:
+  struct State {
+    State(Stats::GaugeSharedPtr gauge, uint64_t value) : gauge_(std::move(gauge)), value_(value) {}
+
+    Stats::GaugeSharedPtr gauge_;
+    uint64_t value_;
+  };
+
+  Stats::ScopeSharedPtr scope_;
+
+  // The map key holds a raw pointer to the gauge. The value holds a ref-counted pointer to ensure
+  // the gauge is not destroyed if it is evicted from the stats scope.
+  absl::flat_hash_map<Stats::Gauge*, State> inflight_gauges_;
+};
+
 Formatter::FormatterProviderPtr
 parseValueFormat(absl::string_view format,
                  const std::vector<Formatter::CommandParserPtr>& commands) {
@@ -46,7 +113,7 @@ StatsAccessLog::StatsAccessLog(const envoy::extensions::access_loggers::stats::v
                                AccessLog::FilterPtr&& filter,
                                const std::vector<Formatter::CommandParserPtr>& commands)
     : Common::ImplBase(std::move(filter)),
-      scope_(context.statsScope().createScope(config.stat_prefix(), true /* evictable */)),
+      scope_(context.statsScope().createScope(config.stat_prefix())),
       stat_name_pool_(scope_->symbolTable()), histograms_([&]() {
         std::vector<Histogram> histograms;
         for (const auto& hist_cfg : config.histograms()) {
@@ -76,6 +143,67 @@ StatsAccessLog::StatsAccessLog(const envoy::extensions::access_loggers::stats::v
           }
         }
         return counters;
+      }()),
+      gauges_([&]() {
+        std::vector<Gauge> gauges;
+        for (const auto& gauge_cfg : config.gauges()) {
+          if (gauge_cfg.has_add_subtract() && gauge_cfg.has_set()) {
+            throw EnvoyException(
+                "Stats logger gauge cannot have both SET and PAIRED_ADD/PAIRED_SUBTRACT "
+                "operations.");
+          }
+
+          if (!gauge_cfg.has_add_subtract() && !gauge_cfg.has_set()) {
+            throw EnvoyException("Stats logger gauge must have at least one operation configured.");
+          }
+
+          absl::InlinedVector<
+              std::pair<envoy::data::accesslog::v3::AccessLogType, Gauge::OperationType>, 2>
+              operations;
+
+          if (gauge_cfg.has_add_subtract()) {
+            if (gauge_cfg.add_subtract().add_log_type() ==
+                    envoy::data::accesslog::v3::AccessLogType::NotSet ||
+                gauge_cfg.add_subtract().sub_log_type() ==
+                    envoy::data::accesslog::v3::AccessLogType::NotSet) {
+              throw EnvoyException(
+                  "Stats logger gauge add/subtract operation must have a valid log type.");
+            }
+            if (gauge_cfg.add_subtract().add_log_type() ==
+                gauge_cfg.add_subtract().sub_log_type()) {
+              throw EnvoyException(
+                  fmt::format("Duplicate access log type '{}' in gauge operations.",
+                              static_cast<int>(gauge_cfg.add_subtract().add_log_type())));
+            }
+            operations.emplace_back(gauge_cfg.add_subtract().add_log_type(),
+                                    Gauge::OperationType::PAIRED_ADD);
+            operations.emplace_back(gauge_cfg.add_subtract().sub_log_type(),
+                                    Gauge::OperationType::PAIRED_SUBTRACT);
+          } else {
+            if (gauge_cfg.set().log_type() == envoy::data::accesslog::v3::AccessLogType::NotSet) {
+              throw EnvoyException("Stats logger gauge set operation must have a valid log type.");
+            }
+            operations.emplace_back(gauge_cfg.set().log_type(), Gauge::OperationType::SET);
+          }
+
+          Gauge& inserted =
+              gauges.emplace_back(NameAndTags(gauge_cfg.stat(), stat_name_pool_, commands), nullptr,
+                                  0, std::move(operations));
+
+          if (!gauge_cfg.value_format().empty() && gauge_cfg.has_value_fixed()) {
+            throw EnvoyException(
+                "Stats logger cannot have both `value_format` and `value_fixed` configured.");
+          }
+          if (!gauge_cfg.value_format().empty()) {
+            inserted.value_formatter_ = parseValueFormat(gauge_cfg.value_format(), commands);
+          } else if (gauge_cfg.has_value_fixed()) {
+            inserted.value_fixed_ = gauge_cfg.value_fixed().value();
+          } else {
+            throw EnvoyException(
+                "Stats logger gauge must have either `value_format` or `value_fixed`.");
+          }
+        }
+        return gauges;
       }()) {}
 
 StatsAccessLog::NameAndTags::NameAndTags(
@@ -182,6 +310,65 @@ void StatsAccessLog::emitLogConst(const Formatter::Context& context,
     auto& counter_stat = scope_->counterFromStatNameWithTags(counter.stat_.name_, tags);
     counter_stat.add(value);
   }
+
+  for (const auto& gauge : gauges_) {
+    emitLogForGauge(gauge, context, stream_info);
+  }
+}
+
+void StatsAccessLog::emitLogForGauge(const Gauge& gauge, const Formatter::Context& context,
+                                     const StreamInfo::StreamInfo& stream_info) const {
+  auto it = std::find_if(gauge.operations_.begin(), gauge.operations_.end(),
+                         [&](const auto& op) { return op.first == context.accessLogType(); });
+  if (it == gauge.operations_.end()) {
+    return;
+  }
+
+  uint64_t value;
+  if (gauge.value_formatter_ != nullptr) {
+    absl::optional<uint64_t> computed_value_opt =
+        getFormatValue(*gauge.value_formatter_, context, stream_info, false);
+    if (!computed_value_opt.has_value()) {
+      return;
+    }
+
+    value = *computed_value_opt;
+  } else {
+    value = gauge.value_fixed_;
+  }
+
+  Gauge::OperationType op = it->second;
+
+  auto [tags, storage] = gauge.stat_.tags(context, stream_info, *scope_);
+  Stats::Gauge::ImportMode import_mode = op == Gauge::OperationType::SET
+                                             ? Stats::Gauge::ImportMode::NeverImport
+                                             : Stats::Gauge::ImportMode::Accumulate;
+  auto& gauge_stat = scope_->gaugeFromStatNameWithTags(gauge.stat_.name_, tags, import_mode);
+
+  if (op == Gauge::OperationType::PAIRED_ADD || op == Gauge::OperationType::PAIRED_SUBTRACT) {
+    auto& filter_state = const_cast<StreamInfo::FilterState&>(stream_info.filterState());
+    if (!filter_state.hasData<AccessLogState>(AccessLogState::key())) {
+      filter_state.setData(AccessLogState::key(), std::make_shared<AccessLogState>(scope_),
+                           StreamInfo::FilterState::StateType::Mutable,
+                           StreamInfo::FilterState::LifeSpan::Request);
+    }
+    auto* state = filter_state.getDataMutable<AccessLogState>(AccessLogState::key());
+
+    if (op == Gauge::OperationType::PAIRED_ADD) {
+      state->addInflightGauge(&gauge_stat, value);
+      gauge_stat.add(value);
+    } else {
+      absl::optional<uint64_t> added_value = state->removeInflightGauge(&gauge_stat);
+      if (added_value.has_value()) {
+        gauge_stat.sub(added_value.value());
+      }
+    }
+    return;
+  }
+
+  if (op == Gauge::OperationType::SET) {
+    gauge_stat.set(value);
+  }
 }
 
 } // namespace StatsAccessLog

source/extensions/access_loggers/stats/stats.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "envoy/data/accesslog/v3/accesslog.pb.h"
 #include "envoy/extensions/access_loggers/stats/v3/stats.pb.h"
 #include "envoy/stats/tag.h"
 
@@ -65,11 +66,29 @@ class StatsAccessLog : public Common::ImplBase {
     uint64_t value_fixed_;
   };
 
+  struct Gauge {
+    enum class OperationType {
+      SET,
+      PAIRED_ADD,
+      PAIRED_SUBTRACT,
+    };
+
+    NameAndTags stat_;
+    Formatter::FormatterProviderPtr value_formatter_;
+    uint64_t value_fixed_;
+    absl::InlinedVector<std::pair<envoy::data::accesslog::v3::AccessLogType, OperationType>, 2>
+        operations_;
+  };
+
+  void emitLogForGauge(const Gauge& gauge, const Formatter::Context& context,
+                       const StreamInfo::StreamInfo& stream_info) const;
+
   const Stats::ScopeSharedPtr scope_;
   Stats::StatNamePool stat_name_pool_;
 
   const std::vector<Histogram> histograms_;
   const std::vector<Counter> counters_;
+  const std::vector<Gauge> gauges_;
 };
 
 } // namespace StatsAccessLog