Add allowlisted metadata-only quota telemetry

zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md

@@ -1923,6 +1923,14 @@ options are used to configure the [AdminServer](#sc_adminserver).
     The URL for listing and issuing commands relative to the
     root URL.  Defaults to "/commands".
 
+* *zookeeper.quotaStats.allowedNamespaces* :
+    (Java system property)
+    A JSON array of exact namespace paths that may be queried by
+    [quota_stats](#sc_quota_stats). Defaults to `[]`, which denies every
+    namespace. Keep this list empty until the namespaces and operational
+    access have been explicitly approved. This is a metadata-disclosure
+    allowlist, not a quota limit or an authentication mechanism.
+
 ### Metrics Providers
 
 **New in 3.6.0:** The following options are used to configure metrics.
@@ -2374,6 +2382,12 @@ Available commands include:
     Reset all observer connection statistics. Companion command to *observers*.
     No new fields returned.
 
+* *quota_stats* :
+    Read only the existing quota usage and limit metadata for one explicitly
+    allowlisted namespace. Requires a `path` parameter. See
+    [bounded quota telemetry](#sc_quota_stats) for configuration, schema,
+    unavailable results and sampling limitations.
+
 * *ruok* :
     No-op command, check if the server is running.
     A response does not necessarily indicate that the
@@ -2439,6 +2453,92 @@ Available commands include:
     Peers can be in one of these phases: ELECTION, DISCOVERY, SYNCHRONIZATION, BROADCAST.
     Returns fields "voting" and "zabstate".
 
+<a name="sc_quota_stats"></a>
+
+##### Bounded quota telemetry
+
+`quota_stats?path=/example-quota` runs through the existing AdminServer
+command URL, for example `/commands/quota_stats?path=/example-quota`.
+`/example-quota` is illustrative, not an approved namespace. Use the existing
+protected administrative access path; this command adds no authentication,
+ACL policy, transport listener or four-letter command.
+
+The JVM property `zookeeper.quotaStats.allowedNamespaces` must contain one
+complete JSON array of strings, for example `["/example-quota"]`. An unset
+property means `[]` and rejects all requests. The command reads the property
+once per request and validates the entire array before reading quota metadata;
+an invalid later entry cannot be hidden by an earlier match. Non-array JSON,
+non-string entries, malformed JSON, trailing JSON values, invalid paths and
+unreadable configuration produce a normal command error rather than permissive
+fallback. Duplicate valid entries have no additional effect.
+
+Both configured and requested paths must pass ZooKeeper's existing znode path
+validation. Root `/`, `/zookeeper` and its descendants are excluded;
+`/zookeeper-client` is not excluded by that reserved-path rule. Paths are
+matched exactly, with no trimming, normalization, ancestor inheritance,
+prefix matching or wildcard expansion. A literal `*` in a valid znode name is
+just a character, not a pattern. A child namespace needs its own explicit
+allowlist entry and its own quota metadata. Encode the `path` query parameter
+normally when its characters require URL encoding.
+
+Invalid/missing paths, invalid configuration and disallowed paths return only
+the usual `command` and non-null `error` fields. An uninitialized or stopped
+server is rejected by the existing command-framework availability guard.
+Successful requests, including unavailable quota samples, have this schema:
+
+| Field | Type | Meaning |
+| --- | --- | --- |
+| `command` | string | `quota_stats` |
+| `error` | null | Request/configuration validation succeeded |
+| `schema_version` | integer | `1` |
+| `path` | string | The exact requested, allowlisted namespace |
+| `count_used` | integer or null | Existing quota-stat node count, including the namespace itself |
+| `bytes_used` | integer or null | Existing quota-stat total znode payload bytes, not path, ACL, packet or JVM-memory bytes |
+| `count_limit` | integer or null | Existing quota count limit; native `-1` is reported as null |
+| `bytes_limit` | integer or null | Existing quota byte limit; native `-1` is reported as null |
+| `available` | boolean | Both exact metadata records are present and valid in this sample |
+| `reason` | string or null | Unavailability reason below, or null when available |
+
+Counts use the native signed 32-bit integer range; byte values use signed
+64-bit integers. Metadata must use the native `count=<int>,bytes=<long>`
+format with exact field names/order and decimal digits. Usage must be
+non-negative; limits must be at least `-1`. Null/empty payloads, malformed
+records and numeric overflow are invalid, not zero usage. An available
+sample can have one or both limits null (unset/unlimited in native quota
+metadata); null is not approved infinite headroom. Zero limits remain zero.
+The command computes no utilization ratios or headroom, and clients must
+not divide by zero or treat unknown values as an approved budget.
+
+An unavailable sample sets **all four numeric fields to null** and uses one
+of these reasons:
+
+| `reason` | Meaning |
+| --- | --- |
+| `namespace_missing` | The namespace does not exist when sampling starts, even if orphan quota metadata remains |
+| `quota_missing` | Both exact quota metadata nodes are absent; an ancestor's quota is not substituted |
+| `quota_incomplete` | Exactly one of the stat/limit nodes is absent |
+| `invalid_quota_stats` | The stat record is null, malformed, negative or out of range |
+| `invalid_quota_limits` | The limit record is null, malformed or out of range |
+| `quota_changed` | Removal/replacement of the namespace or a sampled metadata node was detected during sampling |
+
+The only payloads read are the exact
+`/zookeeper/quota<path>/zookeeper_stats` and
+`/zookeeper/quota<path>/zookeeper_limits` records. Namespace existence and
+node-identity checks use a bounded number of direct lookups. The command
+does not walk the live subtree, search ancestor quotas, create watches,
+repair accounting, alter enforcement or register per-path metrics.
+
+**This is not an atomic or historical snapshot.** Stat and limit records are
+sampled independently with no tree-wide lock; they can reflect different
+instants and existing quota-accounting lag. Identity rechecks detect some
+removals/replacements, not every concurrent change. `available=true` is a
+metadata-availability statement, not a freshness, consistency or safety
+guarantee. Missing or invalid data must remain unknown.
+
+Continue using `monitor/mntr` for existing response-size, session/connection,
+latency and queue metrics, and `watch_summary/wchs` for watch totals.
+`quota_stats` does not duplicate those metrics or introduce namespace labels.
+
 
 <a name="sc_dataFileManagement"></a>
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java

@@ -23,6 +23,7 @@
 import java.io.PrintWriter;
 import java.io.Serializable;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -37,6 +38,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.regex.Pattern;
 import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.InputArchive;
@@ -106,6 +108,8 @@ public class DataTree {
 
     private static final Logger LOG = LoggerFactory.getLogger(DataTree.class);
 
+    private static final Pattern QUOTA_VALUE = Pattern.compile("count=-?[0-9]+,bytes=-?[0-9]+");
+
     private final RateLogger RATE_LOGGER = new RateLogger(LOG, 15 * 60 * 1000);
 
     /**
@@ -760,6 +764,100 @@ public String getMaxPrefixWithQuota(String path) {
         }
     }
 
+    /**
+     * Samples only the exact namespace's existing quota metadata, without traversing
+     * or updating the tree. The two metadata reads are not an atomic snapshot.
+     *
+     * @param path a validated absolute, non-root, non-reserved namespace path
+     */
+    public QuotaStats getQuotaStats(String path) {
+        DataNode namespace = getNode(path);
+        if (namespace == null) {
+            return new QuotaStats(null, null, "namespace_missing");
+        }
+        String statPath = Quotas.statPath(path);
+        String limitPath = Quotas.quotaPath(path);
+        DataNode statNode = getNode(statPath);
+        DataNode limitNode = getNode(limitPath);
+        if (statNode == null && limitNode == null) {
+            return new QuotaStats(null, null, "quota_missing");
+        }
+        if (statNode == null || limitNode == null) {
+            return new QuotaStats(null, null, "quota_incomplete");
+        }
+        StatsTrack usage = readQuotaMetadata(statNode, 0);
+        StatsTrack limits = readQuotaMetadata(limitNode, -1);
+        if (getNode(path) != namespace || getNode(statPath) != statNode || getNode(limitPath) != limitNode) {
+            return new QuotaStats(null, null, "quota_changed");
+        }
+        if (usage == null) {
+            return new QuotaStats(null, null, "invalid_quota_stats");
+        }
+        if (limits == null) {
+            return new QuotaStats(null, null, "invalid_quota_limits");
+        }
+        return new QuotaStats(usage, limits, null);
+    }
+
+    private static StatsTrack readQuotaMetadata(DataNode node, int minimum) {
+        byte[] data = node.getData();
+        if (data == null) {
+            return null;
+        }
+        String value = new String(data, StandardCharsets.UTF_8);
+        // StatsTrack accepts arbitrary field names; validate the metadata format first.
+        if (!QUOTA_VALUE.matcher(value).matches()) {
+            return null;
+        }
+        try {
+            StatsTrack stats = new StatsTrack(value);
+            return stats.getCount() < minimum || stats.getBytes() < minimum ? null : stats;
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
+    /**
+     * A quota metadata sample; unavailable samples expose no numeric values.
+     */
+    public static final class QuotaStats {
+
+        private final StatsTrack usage;
+        private final StatsTrack limits;
+        private final String reason;
+
+        private QuotaStats(StatsTrack usage, StatsTrack limits, String reason) {
+            this.usage = usage;
+            this.limits = limits;
+            this.reason = reason;
+        }
+
+        public Integer getCountUsed() {
+            return usage == null ? null : usage.getCount();
+        }
+
+        public Long getBytesUsed() {
+            return usage == null ? null : usage.getBytes();
+        }
+
+        public Integer getCountLimit() {
+            return limits == null || limits.getCount() == -1 ? null : limits.getCount();
+        }
+
+        public Long getBytesLimit() {
+            return limits == null || limits.getBytes() == -1 ? null : limits.getBytes();
+        }
+
+        public boolean isAvailable() {
+            return reason == null;
+        }
+
+        public String getReason() {
+            return reason;
+        }
+
+    }
+
     public void addWatch(String basePath, Watcher watcher, int mode) {
         WatcherMode watcherMode = WatcherMode.fromZooDef(mode);
         dataWatches.addWatch(basePath, watcher, watcherMode);

zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java

@@ -19,7 +19,11 @@
 package org.apache.zookeeper.server.admin;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
 import java.util.Collections;
@@ -34,8 +38,11 @@
 import java.util.stream.Collectors;
 import org.apache.zookeeper.Environment;
 import org.apache.zookeeper.Environment.Entry;
+import org.apache.zookeeper.Quotas;
 import org.apache.zookeeper.Version;
+import org.apache.zookeeper.common.PathUtils;
 import org.apache.zookeeper.server.DataTree;
+import org.apache.zookeeper.server.DataTree.QuotaStats;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ZooKeeperServer;
@@ -142,6 +149,7 @@ public static Command getCommand(String cmdName) {
         registerCommand(new LeaderCommand());
         registerCommand(new MonitorCommand());
         registerCommand(new ObserverCnxnStatResetCommand());
+        registerCommand(new QuotaStatsCommand());
         registerCommand(new RuokCommand());
         registerCommand(new SetTraceMaskCommand());
         registerCommand(new SrvrCommand());
@@ -480,6 +488,81 @@ public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs)
 
     }
 
+    /**
+     * Samples the existing quota metadata for one explicitly allowlisted namespace.
+     */
+    public static class QuotaStatsCommand extends CommandBase {
+
+        private static final String ALLOWED_NAMESPACES = "zookeeper.quotaStats.allowedNamespaces";
+        private static final JsonFactory JSON = new JsonFactory();
+
+        public QuotaStatsCommand() {
+            super(Collections.singletonList("quota_stats"));
+        }
+
+        @Override
+        public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs) {
+            String path = kwargs == null ? null : kwargs.get("path");
+            if (!isValidNamespacePath(path)) {
+                return new CommandResponse(getPrimaryName(), "quota_stats requires a valid namespace path");
+            }
+            final boolean allowed;
+            try {
+                allowed = isAllowed(path, System.getProperty(ALLOWED_NAMESPACES, "[]"));
+            } catch (IOException | IllegalArgumentException | SecurityException e) {
+                return new CommandResponse(getPrimaryName(), "Invalid " + ALLOWED_NAMESPACES
+                                           + ": expected a JSON array of valid namespace paths");
+            }
+            if (!allowed) {
+                return new CommandResponse(getPrimaryName(), "Path is not allowlisted for quota_stats");
+            }
+
+            QuotaStats stats = zkServer.getZKDatabase().getDataTree().getQuotaStats(path);
+            CommandResponse response = initializeResponse();
+            response.put("schema_version", 1);
+            response.put("path", path);
+            response.put("count_used", stats.getCountUsed());
+            response.put("bytes_used", stats.getBytesUsed());
+            response.put("count_limit", stats.getCountLimit());
+            response.put("bytes_limit", stats.getBytesLimit());
+            response.put("available", stats.isAvailable());
+            response.put("reason", stats.getReason());
+            return response;
+        }
+
+        private static boolean isValidNamespacePath(String path) {
+            try {
+                PathUtils.validatePath(path);
+            } catch (IllegalArgumentException e) {
+                return false;
+            }
+            return !"/".equals(path)
+                   && !Quotas.procZookeeper.equals(path)
+                   && !path.startsWith(Quotas.procZookeeper + "/");
+        }
+
+        private static boolean isAllowed(String path, String configuration) throws IOException {
+            boolean allowed = false;
+            try (JsonParser parser = JSON.createParser(configuration)) {
+                if (parser.nextToken() != JsonToken.START_ARRAY) {
+                    throw new IllegalArgumentException();
+                }
+                JsonToken token;
+                while ((token = parser.nextToken()) != JsonToken.END_ARRAY) {
+                    if (token != JsonToken.VALUE_STRING || !isValidNamespacePath(parser.getText())) {
+                        throw new IllegalArgumentException();
+                    }
+                    allowed |= path.equals(parser.getText());
+                }
+                if (parser.nextToken() != null) {
+                    throw new IllegalArgumentException();
+                }
+            }
+            return allowed;
+        }
+
+    }
+
     /**
      * No-op command, check if the server is running
      */