This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 533096bd43 feat(logging): expose queue admission and capacity metrics
(#7833)
533096bd43 is described below
commit 533096bd43621c5ef4194193da0363aea1c8ffcc
Author: Duansg <[email protected]>
AuthorDate: Sun Oct 18 16:35:41 2026 -0700
feat(logging): expose queue admission and capacity metrics (#7833)
---
.../shenyu/common/metrics/LoggingQueueMetrics.java | 163 ++++++++++
.../common/metrics/LoggingQueueMetricsTest.java | 47 +++
.../shenyu-plugin-logging-common/README.md | 62 ++++
.../common/collector/AbstractLogCollector.java | 33 +-
.../common/collector/LogCollectorMetricsTest.java | 341 +++++++++++++++++++++
.../shenyu/plugin/metrics/constant/LabelNames.java | 12 +
.../prometheus/LoggingQueueMetricsCollector.java | 66 ++++
.../prometheus/PrometheusMetricsService.java | 1 +
.../LoggingQueueMetricsCollectorTest.java | 97 ++++++
.../prometheus/PrometheusMetricsServiceTest.java | 24 ++
10 files changed, 841 insertions(+), 5 deletions(-)
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/metrics/LoggingQueueMetrics.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/metrics/LoggingQueueMetrics.java
new file mode 100644
index 0000000000..1f750396a4
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/metrics/LoggingQueueMetrics.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.common.metrics;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.LongSupplier;
+
+/**
+ * Admission counters and live queue measurements, independent of the metrics
backend.
+ * Only collector implementation classes and the two queue modes identify a
series.
+ * Suppliers are evaluated when metrics are scraped, never when a record is
admitted.
+ */
+public final class LoggingQueueMetrics {
+
+ private static final Map<String, LoggingQueueMetrics> REGISTERED = new
ConcurrentHashMap<>();
+
+ private final String collector;
+
+ private final Mode mode;
+
+ private final LongSupplier queuedRecords;
+
+ private final LongSupplier capacity;
+
+ private final LongAdder accepted = new LongAdder();
+
+ private final LongAdder dropped = new LongAdder();
+
+ /**
+ * Create measurements for one collector and queue mode.
+ *
+ * @param collectorType implementation class, never a request or selector
identifier
+ * @param mode queue mode
+ * @param queuedRecords current number of queued records
+ * @param capacity current admission capacity, aggregated across selector
queues
+ */
+ public LoggingQueueMetrics(final Class<?> collectorType, final Mode mode,
+ final LongSupplier queuedRecords, final
LongSupplier capacity) {
+ this.collector = collectorType.getName();
+ this.mode = mode;
+ this.queuedRecords = queuedRecords;
+ this.capacity = capacity;
+ }
+
+ /**
+ * Publish measurements after the collector starts.
+ */
+ public void register() {
+ REGISTERED.put(collector + mode.name(), this);
+ }
+
+ /**
+ * Remove only this collector's measurements when it stops.
+ */
+ public void unregister() {
+ REGISTERED.remove(collector + mode.name(), this);
+ }
+
+ /**
+ * Get the currently running collectors, including collectors started
before metrics.
+ *
+ * @return snapshot of registered measurements
+ */
+ public static Collection<LoggingQueueMetrics> registered() {
+ return List.copyOf(REGISTERED.values());
+ }
+
+ /**
+ * Record an attempted admission. Call only after lifecycle and client
checks.
+ *
+ * @param admitted whether the record was queued; false means an admission
rejection
+ */
+ public void recordAdmission(final boolean admitted) {
+ if (admitted) {
+ accepted.increment();
+ } else {
+ dropped.increment();
+ }
+ }
+
+ /**
+ * Get the collector implementation name.
+ *
+ * @return bounded collector label
+ */
+ public String getCollector() {
+ return collector;
+ }
+
+ /**
+ * Get the queue mode.
+ *
+ * @return queue mode
+ */
+ public Mode getMode() {
+ return mode;
+ }
+
+ /**
+ * Get admissions over this collector instance's lifetime, including
restarts.
+ *
+ * @return admitted records, not successful backend sends
+ */
+ public long getAccepted() {
+ return accepted.sum();
+ }
+
+ /**
+ * Get queue admission rejections, excluding sampling and backend send
failures.
+ *
+ * @return rejected records
+ */
+ public long getDropped() {
+ return dropped.sum();
+ }
+
+ /**
+ * Get the current backlog, excluding batches already drained for sending.
+ *
+ * @return queued records
+ */
+ public long getQueuedRecords() {
+ return queuedRecords.getAsLong();
+ }
+
+ /**
+ * Get effective admission capacity; a backlog retained after shrinking
may exceed it.
+ *
+ * @return capacity in records
+ */
+ public long getCapacity() {
+ return capacity.getAsLong();
+ }
+
+ /**
+ * Fixed queue modes, independent of request and selector data.
+ */
+ public enum Mode {
+ /** Shared queue for a single client. */
+ SINGLE,
+ /** Aggregated queues for selector-specific clients. */
+ MULTI
+ }
+}
diff --git
a/shenyu-common/src/test/java/org/apache/shenyu/common/metrics/LoggingQueueMetricsTest.java
b/shenyu-common/src/test/java/org/apache/shenyu/common/metrics/LoggingQueueMetricsTest.java
new file mode 100644
index 0000000000..9d96c576d8
--- /dev/null
+++
b/shenyu-common/src/test/java/org/apache/shenyu/common/metrics/LoggingQueueMetricsTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.common.metrics;
+
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics.Mode;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verify replacement collectors cannot be removed by an older instance
closing.
+ */
+class LoggingQueueMetricsTest {
+
+ @Test
+ void unregisterOnlyRemovesTheOwningCollector() {
+ final LoggingQueueMetrics previous = new
LoggingQueueMetrics(getClass(), Mode.SINGLE, () -> 0, () -> 1);
+ final LoggingQueueMetrics current = new
LoggingQueueMetrics(getClass(), Mode.SINGLE, () -> 0, () -> 2);
+ try {
+ previous.register();
+ current.register();
+ previous.unregister();
+ assertFalse(LoggingQueueMetrics.registered().contains(previous));
+ assertTrue(LoggingQueueMetrics.registered().contains(current));
+ } finally {
+ previous.unregister();
+ current.unregister();
+ }
+ assertFalse(LoggingQueueMetrics.registered().contains(current));
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/README.md
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/README.md
new file mode 100644
index 0000000000..5a6cf01909
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/README.md
@@ -0,0 +1,62 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+# Logging queue metrics
+
+When ShenYu's metrics starter is enabled (`shenyu.metrics.enabled=true`),
+the existing Prometheus endpoint exposes these metrics for running collectors
+that extend `AbstractLogCollector` (Kafka, Elasticsearch and RocketMQ):
+
+| Metric | Type | Meaning |
+| --- | --- | --- |
+| `shenyu_logging_queue_accepted_total` | Counter | Records successfully
admitted to a logging queue. This is not a backend delivery count. |
+| `shenyu_logging_queue_dropped_total` | Counter | Records rejected at queue
admission because the queue has reached its capacity. |
+| `shenyu_logging_queue_size` | Gauge | Records currently in the queues,
excluding batches already drained for sending. |
+| `shenyu_logging_queue_capacity` | Gauge | Effective admission capacity in
records, summed across the queues in the indicated mode. |
+
+The only labels are `collector` (the fully qualified collector implementation
+class) and `mode` (`single` or `multi`). Selector IDs, paths, request IDs and
+backend addresses are never labels. In multi-client mode, measurements
aggregate
+all selector queues for the collector. Consequently, an individual selector
+queue can overflow even when the aggregate occupancy is below aggregate
capacity;
+use the dropped counter to alert on any admission loss.
+
+For single-client mode, capacity is the live configured `bufferQueueSize`, not
+the physical capacity of the underlying unbounded deque. After shrinking the
+configuration, occupancy may temporarily exceed capacity while the existing
+backlog drains. For multi-client mode, capacity is the sum of the actual bounds
+chosen when each selector queue was created. Existing queues retain their
bounds
+after configuration changes; new queues use the latest configuration. Before
the
+first selector queue is created, aggregate capacity is zero. Inactive modes
have
+zero admission capacity but still expose any backlog retained after a mode
switch.
+
+Only attempted admissions are counted. Rule sampling, null logs, collectors
that
+have not started or have stopped, and missing clients do not increase the
dropped
+counter. Backend send failures are also excluded: a record already accepted can
+still fail during delivery.
+
+Counters are retained across stop/start cycles of the same collector instance.
+Stopped collectors are removed from the exported series, and a new collector
+instance or gateway process starts with new counters. Metrics work regardless
of
+whether the logging collector or the metrics service starts first. Gauges read
+live queue state at scrape time, so idle collectors do not retain stale
occupancy
+values after draining. Concurrent scrapes are observations rather than atomic
+snapshots across all queues and counters.
+
+Admission uses the existing non-waiting capacity checks and `offer` behavior.
+Instrumentation adds only thread-safe counters to the request path; it adds no
+worker, auxiliary record queue or backend metrics dependency to logging
plugins.
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
index 3576996a76..77a0b9880e 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
@@ -23,6 +23,8 @@ import
org.apache.shenyu.common.concurrent.ShenyuThreadFactory;
import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
import org.apache.shenyu.common.config.ShenyuConfig;
import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics.Mode;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.common.utils.ThreadUtils;
import org.apache.shenyu.plugin.logging.common.client.AbstractLogConsumeClient;
@@ -44,6 +46,7 @@ import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import static
org.apache.shenyu.plugin.logging.desensitize.api.utils.DataDesensitizeUtils.desensitizeForBody;
import static
org.apache.shenyu.plugin.logging.desensitize.api.utils.DataDesensitizeUtils.desensitizeForSingleWord;
@@ -69,6 +72,16 @@ public abstract class AbstractLogCollector<T extends
AbstractLogConsumeClient<?,
private final AtomicBoolean started = new AtomicBoolean(false);
+ private final AtomicLong multiBufferCapacity = new AtomicLong();
+
+ private final LoggingQueueMetrics singleClientMetrics = new
LoggingQueueMetrics(getClass(), Mode.SINGLE,
+ () -> started.get() ? bufferQueue.size() : 0,
+ () -> isActiveMode(false) ? bufferSize : 0);
+
+ private final LoggingQueueMetrics multiClientMetrics = new
LoggingQueueMetrics(getClass(), Mode.MULTI,
+ () -> started.get() ?
bufferQueueS.values().stream().mapToLong(BlockingQueue::size).sum() : 0,
+ () -> isActiveMode(true) ? multiBufferCapacity.get() : 0);
+
private ShenyuThreadPoolExecutor executor;
@Override
@@ -95,6 +108,12 @@ public abstract class AbstractLogCollector<T extends
AbstractLogConsumeClient<?,
threadExecutor.shutdownNow();
throw e;
}
+ singleClientMetrics.register();
+ multiClientMetrics.register();
+ }
+
+ private boolean isActiveMode(final boolean multiClient) {
+ return started.get() && getMultiClient() == multiClient;
}
private void refreshBufferSize() {
@@ -113,15 +132,15 @@ public abstract class AbstractLogCollector<T extends
AbstractLogConsumeClient<?,
if (getMultiClient()) {
String selectorId = log.getSelectorId();
BlockingQueue<L> bufferQueue =
bufferQueueS.computeIfAbsent(selectorId, bufferQueueS -> initQueue(selectorId));
- bufferQueue.offer(log);
+ multiClientMetrics.recordAdmission(bufferQueue.offer(log));
} else {
BlockingQueue<L> queue = bufferQueue;
+ boolean admitted;
synchronized (queue) {
// On shrink, retain the backlog but reject new logs until it
falls below the new limit.
- if (queue.size() < bufferSize) {
- queue.offer(log);
- }
+ admitted = queue.size() < bufferSize && queue.offer(log);
}
+ singleClientMetrics.recordAdmission(admitted);
}
}
@@ -165,8 +184,10 @@ public abstract class AbstractLogCollector<T extends
AbstractLogConsumeClient<?,
}
private BlockingQueue<L> initQueue(final String selectorId) {
- BlockingQueue<L> queue = new
LinkedBlockingDeque<>(getLogCollectConfig().getBufferQueueSize());
+ int capacity = getLogCollectConfig().getBufferQueueSize();
+ BlockingQueue<L> queue = new LinkedBlockingDeque<>(capacity);
lastPushTimeS.put(selectorId, System.currentTimeMillis());
+ multiBufferCapacity.addAndGet(capacity);
return queue;
}
@@ -320,6 +341,8 @@ public abstract class AbstractLogCollector<T extends
AbstractLogConsumeClient<?,
try {
flushBufferQueues();
} finally {
+ singleClientMetrics.unregister();
+ multiClientMetrics.unregister();
if (Objects.nonNull(logCollectClient)) {
logCollectClient.close();
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/LogCollectorMetricsTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/LogCollectorMetricsTest.java
new file mode 100644
index 0000000000..bf0b9f1654
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/LogCollectorMetricsTest.java
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.logging.common.collector;
+
+import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.enums.PluginEnum;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics.Mode;
+import org.apache.shenyu.plugin.logging.common.AbstractLoggingPlugin;
+import org.apache.shenyu.plugin.logging.common.client.AbstractLogConsumeClient;
+import org.apache.shenyu.plugin.logging.common.config.GenericApiConfig;
+import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
+import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
+import
org.apache.shenyu.plugin.logging.common.handler.AbstractLogPluginDataHandler;
+import org.apache.shenyu.plugin.logging.common.sampler.Sampler;
+import org.apache.shenyu.plugin.logging.desensitize.api.matcher.KeyWordMatch;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedConstruction;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.BlockingQueue;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Verify admission accounting against real queues with the consumer held idle.
+ */
+class LogCollectorMetricsTest {
+
+ private TestCollector collector;
+
+ private MockedConstruction<ShenyuThreadPoolExecutor> executors;
+
+ @BeforeEach
+ void setUp() {
+ executors = mockConstruction(ShenyuThreadPoolExecutor.class);
+ collector = new TestCollector();
+ collector.config.setBufferQueueSize(1);
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ try {
+ reset(collector.client);
+ collector.close();
+ } finally {
+ executors.close();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void countsOnlyQueueAdmissionRejections(final boolean multi) throws
Exception {
+ collector.multi = multi;
+ collector.collect(log("selector"));
+ collector.start();
+ final LoggingQueueMetrics metrics = metrics(multi);
+ assertEquals(0, metrics.getDropped());
+ collector.collect(log("selector"));
+ collector.collect(log("selector"));
+ assertEquals(1, metrics.getAccepted());
+ assertEquals(1, metrics.getDropped());
+ assertEquals(1, metrics.getQueuedRecords());
+ assertEquals(1, metrics.getCapacity());
+ collector.collect(null);
+ collector.clientAvailable = false;
+ collector.collect(log("selector"));
+ assertEquals(1, metrics.getAccepted());
+ assertEquals(1, metrics.getDropped());
+ collector.clientAvailable = true;
+ collector.close();
+ collector.collect(log("selector"));
+ assertEquals(1, metrics.getDropped());
+ assertEquals(0, metrics.getQueuedRecords());
+ assertEquals(0, metrics.getCapacity());
+ assertFalse(LoggingQueueMetrics.registered().contains(metrics));
+ verify(collector.client).consume(anyList());
+ }
+
+ @Test
+ void reflectsLiveSingleClientCapacityWithoutDiscardingBacklog() {
+ collector.start();
+ final LoggingQueueMetrics metrics = metrics(false);
+ collector.collect(log("first"));
+ collector.config.setBufferQueueSize(3);
+ collector.start();
+ collector.collect(log("second"));
+ collector.collect(log("third"));
+ assertEquals(3, metrics.getCapacity());
+ assertEquals(3, metrics.getQueuedRecords());
+ collector.config.setBufferQueueSize(1);
+ collector.start();
+ collector.collect(log("fourth"));
+ assertEquals(3, metrics.getAccepted());
+ assertEquals(1, metrics.getDropped());
+ assertEquals(1, metrics.getCapacity());
+ assertEquals(3, metrics.getQueuedRecords());
+ BlockingQueue<?> queue = (BlockingQueue<?>)
ReflectionTestUtils.getField(collector, "bufferQueue");
+ queue.clear();
+ assertEquals(0, metrics.getQueuedRecords());
+ collector.collect(log("fifth"));
+ assertEquals(4, metrics.getAccepted());
+ assertEquals(1, metrics.getQueuedRecords());
+ }
+
+ @Test
+ void aggregatesSelectorQueuesUsingTheirActualCapacities() {
+ collector.multi = true;
+ collector.start();
+ final LoggingQueueMetrics metrics = metrics(true);
+ assertEquals(0, metrics.getCapacity());
+ collector.collect(log("first"));
+ collector.config.setBufferQueueSize(2);
+ collector.start();
+ collector.collect(log("second"));
+ collector.collect(log("second"));
+ collector.collect(log("first"));
+ collector.collect(log("second"));
+ assertEquals(3, metrics.getAccepted());
+ assertEquals(2, metrics.getDropped());
+ assertEquals(3, metrics.getQueuedRecords());
+ assertEquals(3, metrics.getCapacity());
+ assertEquals(0, metrics(false).getCapacity());
+ collector.processMultiClientBufferQueues(1, 100);
+ assertEquals(1, metrics.getQueuedRecords());
+ assertEquals(3, metrics.getCapacity());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void accountsForConcurrentProducers(final boolean multi) {
+ collector.multi = multi;
+ collector.config.setBufferQueueSize(10);
+ collector.start();
+ IntStream.range(0, 1000).parallel().forEach(index ->
collector.collect(log("selector")));
+ final LoggingQueueMetrics metrics = metrics(multi);
+ assertEquals(10, metrics.getAccepted());
+ assertEquals(990, metrics.getDropped());
+ assertEquals(10, metrics.getQueuedRecords());
+ assertEquals(10, metrics.getCapacity());
+ }
+
+ @Test
+ void backendFailuresAreNotAdmissionDrops() throws Exception {
+ collector.multi = true;
+ collector.start();
+ collector.collect(log("selector"));
+ doThrow(new Exception("backend
unavailable")).when(collector.client).consume(anyList());
+ collector.processMultiClientBufferQueues(1, 100);
+ final LoggingQueueMetrics metrics = metrics(true);
+ assertEquals(1, metrics.getAccepted());
+ assertEquals(0, metrics.getDropped());
+ assertEquals(0, metrics.getQueuedRecords());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void consumingUpdatesOccupancyWithoutChangingAdmissionCounters(final
boolean multi) {
+ collector.multi = multi;
+ collector.config.setBufferQueueSize(2);
+ collector.start();
+ collector.collect(log("selector"));
+ collector.collect(log("selector"));
+ final LoggingQueueMetrics metrics = metrics(multi);
+ if (multi) {
+ collector.processMultiClientBufferQueues(1, 100);
+ } else {
+ BlockingQueue<?> queue = (BlockingQueue<?>)
ReflectionTestUtils.getField(collector, "bufferQueue");
+ ReflectionTestUtils.invokeMethod(collector, "processBufferQueue",
queue, 1, 100, new ArrayList<>(), 0L);
+ }
+ assertEquals(1, metrics.getQueuedRecords());
+ assertEquals(2, metrics.getCapacity());
+ assertEquals(2, metrics.getAccepted());
+ assertEquals(0, metrics.getDropped());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void samplingDoesNotCountAsQueueOverflow(final boolean multi) {
+ collector.multi = multi;
+ collector.start();
+ collector.collect(log("sampled-out-selector"));
+ GenericApiConfig apiConfig = new GenericApiConfig();
+ apiConfig.setSampler(Sampler.NEVER_SAMPLE);
+ SelectorData selector = new SelectorData();
+ selector.setId("sampled-out-selector");
+ RuleData rule = new RuleData();
+ rule.setId("sampled-out-rule");
+ rule.setSelectorId(selector.getId());
+
AbstractLogPluginDataHandler.getSelectApiConfigMap().put(selector.getId(),
apiConfig);
+ try {
+ AtomicBoolean chainExecuted = new AtomicBoolean();
+
loggingPlugin().doExecute(MockServerWebExchange.from(MockServerHttpRequest.get("/sampled-out").build()),
exchange -> {
+ chainExecuted.set(true);
+ return Mono.empty();
+ }, selector, rule).block();
+ assertTrue(chainExecuted.get());
+ assertEquals(1, metrics(multi).getQueuedRecords());
+ assertEquals(1, metrics(multi).getAccepted());
+ assertEquals(0, metrics(multi).getDropped());
+ } finally {
+
AbstractLogPluginDataHandler.getSelectApiConfigMap().remove(selector.getId());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void modeChangesDoNotHideRetainedBacklogs(final boolean multi) {
+ collector.multi = multi;
+ collector.start();
+ collector.collect(log("selector"));
+ final LoggingQueueMetrics metrics = metrics(multi);
+ try {
+ collector.multi = !multi;
+ assertEquals(1, metrics.getQueuedRecords());
+ assertEquals(0, metrics.getCapacity());
+ assertEquals(1, metrics.getAccepted());
+ assertEquals(0, metrics.getDropped());
+ } finally {
+ collector.multi = multi;
+ }
+ assertEquals(1, metrics.getCapacity());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void retainsCountersAndRestoresGaugesOnRestart(final boolean multi) throws
Exception {
+ collector.multi = multi;
+ collector.start();
+ collector.collect(log("selector"));
+ collector.collect(log("selector"));
+ final LoggingQueueMetrics metrics = metrics(multi);
+ collector.close();
+ collector.start();
+ assertSame(metrics, metrics(multi));
+ assertEquals(1, metrics.getDropped());
+ assertEquals(0, metrics.getQueuedRecords());
+ assertEquals(1, metrics.getCapacity());
+ collector.collect(log("selector"));
+ assertEquals(2, metrics.getAccepted());
+ }
+
+ private LoggingQueueMetrics metrics(final boolean multi) {
+ return LoggingQueueMetrics.registered().stream()
+ .filter(metric ->
metric.getCollector().equals(TestCollector.class.getName()))
+ .filter(metric -> metric.getMode() == (multi ? Mode.MULTI :
Mode.SINGLE))
+ .findFirst().orElseThrow(() -> new AssertionError("Collector
did not register queue metrics"));
+ }
+
+ private AbstractLoggingPlugin<ShenyuRequestLog> loggingPlugin() {
+ return new AbstractLoggingPlugin<>() {
+ @Override
+ protected LogCollector<ShenyuRequestLog> logCollector() {
+ return collector;
+ }
+
+ @Override
+ protected PluginEnum pluginEnum() {
+ return PluginEnum.LOGGING_KAFKA;
+ }
+
+ @Override
+ protected ShenyuRequestLog doLogExecute(final ServerWebExchange
exchange, final SelectorData selector, final RuleData rule) {
+ throw new AssertionError("A sampled-out request must not
produce a log");
+ }
+ };
+ }
+
+ private ShenyuRequestLog log(final String selector) {
+ ShenyuRequestLog log = new ShenyuRequestLog();
+ log.setSelectorId(selector);
+ return log;
+ }
+
+ private static final class TestCollector extends
AbstractLogCollector<AbstractLogConsumeClient<?, ShenyuRequestLog>,
ShenyuRequestLog, GenericGlobalConfig> {
+
+ private final GenericGlobalConfig config = new GenericGlobalConfig();
+
+ private final AbstractLogConsumeClient<?, ShenyuRequestLog> client =
mock(AbstractLogConsumeClient.class);
+
+ private boolean multi;
+
+ private boolean clientAvailable = true;
+
+ @Override
+ protected AbstractLogConsumeClient<?, ShenyuRequestLog>
getLogConsumeClient() {
+ return clientAvailable ? client : null;
+ }
+
+ @Override
+ protected GenericGlobalConfig getLogCollectConfig() {
+ return config;
+ }
+
+ @Override
+ protected boolean getMultiClient() {
+ return multi;
+ }
+
+ @Override
+ protected void desensitizeLog(final ShenyuRequestLog log, final
KeyWordMatch match, final String algorithm) {
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
index eccada895e..1f2f16d4da 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/constant/LabelNames.java
@@ -22,6 +22,18 @@ package org.apache.shenyu.plugin.metrics.constant;
*/
public final class LabelNames {
+ /** Records accepted at the logging queue admission boundary, not backend
deliveries. */
+ public static final String LOGGING_QUEUE_ACCEPTED_TOTAL =
"shenyu_logging_queue_accepted_total";
+
+ /** Records rejected at the logging queue admission boundary, not backend
failures. */
+ public static final String LOGGING_QUEUE_DROPPED_TOTAL =
"shenyu_logging_queue_dropped_total";
+
+ /** Queued records, aggregated by collector implementation and queue mode.
*/
+ public static final String LOGGING_QUEUE_SIZE =
"shenyu_logging_queue_size";
+
+ /** Effective admission capacity, aggregated by collector implementation
and queue mode. */
+ public static final String LOGGING_QUEUE_CAPACITY =
"shenyu_logging_queue_capacity";
+
/** Completed authenticated, parsed MCP logical calls, labelled only by
outcome. */
public static final String AGENT_MCP_CALLS_TOTAL =
"shenyu_agent_mcp_calls_total";
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/LoggingQueueMetricsCollector.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/LoggingQueueMetricsCollector.java
new file mode 100644
index 0000000000..7bee9f2932
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/LoggingQueueMetricsCollector.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.metrics.prometheus;
+
+import io.prometheus.client.Collector;
+import io.prometheus.client.CounterMetricFamily;
+import io.prometheus.client.GaugeMetricFamily;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics;
+import org.apache.shenyu.plugin.metrics.constant.LabelNames;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * Scrape logging admission metrics without depending on a logging plugin or
polling thread.
+ */
+public final class LoggingQueueMetricsCollector extends Collector implements
Collector.Describable {
+
+ private static final List<String> LABELS = List.of("collector", "mode");
+
+ @Override
+ public List<MetricFamilySamples> collect() {
+ return samples(LoggingQueueMetrics.registered());
+ }
+
+ @Override
+ public List<MetricFamilySamples> describe() {
+ return samples(Collections.emptyList());
+ }
+
+ private List<MetricFamilySamples> samples(final
Collection<LoggingQueueMetrics> sources) {
+ final CounterMetricFamily accepted = new
CounterMetricFamily(LabelNames.LOGGING_QUEUE_ACCEPTED_TOTAL,
+ "Records admitted to logging queues; does not count successful
backend deliveries", LABELS);
+ final CounterMetricFamily dropped = new
CounterMetricFamily(LabelNames.LOGGING_QUEUE_DROPPED_TOTAL,
+ "Records rejected by logging queue capacity; excludes
sampling, disabled collectors and backend send failures", LABELS);
+ final GaugeMetricFamily size = new
GaugeMetricFamily(LabelNames.LOGGING_QUEUE_SIZE,
+ "Records currently queued, excluding batches already drained
for sending; aggregated across selector queues", LABELS);
+ final GaugeMetricFamily capacity = new
GaugeMetricFamily(LabelNames.LOGGING_QUEUE_CAPACITY,
+ "Effective logging queue admission capacity in records; summed
across selector queues; zero for inactive modes", LABELS);
+ for (LoggingQueueMetrics source : sources) {
+ List<String> labels = List.of(source.getCollector(),
source.getMode().name().toLowerCase(Locale.ROOT));
+ accepted.addMetric(labels, source.getAccepted());
+ dropped.addMetric(labels, source.getDropped());
+ size.addMetric(labels, source.getQueuedRecords());
+ capacity.addMetric(labels, source.getCapacity());
+ }
+ return List.of(accepted, dropped, size, capacity);
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsService.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsService.java
index 7180d47506..3a0dc2d328 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsService.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsService.java
@@ -101,6 +101,7 @@ public final class PrometheusMetricsService implements
MetricsService {
if (!registered.compareAndSet(false, true)) {
return;
}
+ new LoggingQueueMetricsCollector().register();
String jvmEnabled =
String.valueOf(config.getProps().getProperty("jvm_enabled"));
if (StringUtils.isNotEmpty(jvmEnabled)) {
boolean enabled = Boolean.parseBoolean(jvmEnabled);
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/LoggingQueueMetricsCollectorTest.java
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/LoggingQueueMetricsCollectorTest.java
new file mode 100644
index 0000000000..46148068bc
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/LoggingQueueMetricsCollectorTest.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.metrics.prometheus;
+
+import io.prometheus.client.CollectorRegistry;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics.Mode;
+import org.apache.shenyu.plugin.metrics.constant.LabelNames;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Verify exported values, late registration and collector lifecycle with real
scrapes.
+ */
+class LoggingQueueMetricsCollectorTest {
+
+ private final AtomicLong size = new AtomicLong();
+
+ private final AtomicLong capacity = new AtomicLong(10);
+
+ private final LoggingQueueMetrics metrics = new
LoggingQueueMetrics(getClass(), Mode.MULTI, size::get, capacity::get);
+
+ private final CollectorRegistry registry = new CollectorRegistry();
+
+ @AfterEach
+ void tearDown() {
+ metrics.unregister();
+ registry.clear();
+ }
+
+ @Test
+ void exportsExistingAdmissionsAndLiveQueueState() {
+ metrics.register();
+ metrics.recordAdmission(true);
+ metrics.recordAdmission(false);
+ size.set(1);
+ new LoggingQueueMetricsCollector().register(registry);
+ assertEquals(1.0, sample(LabelNames.LOGGING_QUEUE_ACCEPTED_TOTAL));
+ assertEquals(1.0, sample(LabelNames.LOGGING_QUEUE_DROPPED_TOTAL));
+ assertEquals(1.0, sample(LabelNames.LOGGING_QUEUE_SIZE));
+ assertEquals(10.0, sample(LabelNames.LOGGING_QUEUE_CAPACITY));
+ // No new admission: a scrape still observes draining and
configuration changes.
+ size.set(0);
+ capacity.set(20);
+ assertEquals(0.0, sample(LabelNames.LOGGING_QUEUE_SIZE));
+ assertEquals(20.0, sample(LabelNames.LOGGING_QUEUE_CAPACITY));
+ assertEquals(1.0, sample(LabelNames.LOGGING_QUEUE_DROPPED_TOTAL));
+ }
+
+ @Test
+ void discoversCollectorsStartedAfterMetricsAndRemovesStoppedCollectors() {
+ new LoggingQueueMetricsCollector().register(registry);
+ assertNull(sample(LabelNames.LOGGING_QUEUE_SIZE));
+ metrics.register();
+ assertEquals(0.0, sample(LabelNames.LOGGING_QUEUE_DROPPED_TOTAL));
+ assertEquals(10.0, sample(LabelNames.LOGGING_QUEUE_CAPACITY));
+ metrics.recordAdmission(false);
+ metrics.unregister();
+ assertNull(sample(LabelNames.LOGGING_QUEUE_SIZE));
+ assertNull(sample(LabelNames.LOGGING_QUEUE_DROPPED_TOTAL));
+ metrics.register();
+ assertEquals(1.0, sample(LabelNames.LOGGING_QUEUE_DROPPED_TOTAL));
+ }
+
+ @Test
+ void exportsOnlyCollectorAndModeLabels() {
+ metrics.register();
+ new LoggingQueueMetricsCollector().register(registry);
+ registry.metricFamilySamples().asIterator().forEachRemaining(family ->
+ family.samples.forEach(sample ->
assertEquals(List.of("collector", "mode"), sample.labelNames)));
+ }
+
+ private Double sample(final String name) {
+ return registry.getSampleValue(name, new String[]{"collector",
"mode"}, new String[]{getClass().getName(), "multi"});
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsServiceTest.java
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsServiceTest.java
index 4cf59fbb0d..57ff24df36 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsServiceTest.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/prometheus/PrometheusMetricsServiceTest.java
@@ -20,18 +20,24 @@ package org.apache.shenyu.plugin.metrics.prometheus;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.exporter.HTTPServer;
import org.apache.shenyu.common.config.ShenyuConfig.MetricsConfig;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics;
+import org.apache.shenyu.common.metrics.LoggingQueueMetrics.Mode;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.common.utils.ReflectUtils;
+import org.apache.shenyu.plugin.metrics.constant.LabelNames;
import org.hamcrest.MatcherAssert;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
import java.lang.reflect.Field;
+import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -46,6 +52,24 @@ public final class PrometheusMetricsServiceTest {
CollectorRegistry.defaultRegistry.clear();
}
+ @Test
+ public void testRegistersLoggingQueueMetrics() {
+ LoggingQueueMetrics metrics = new LoggingQueueMetrics(getClass(),
Mode.SINGLE, () -> 2, () -> 10);
+ try {
+ metrics.register();
+ MetricsConfig config = new MetricsConfig();
+ config.setProps(new Properties());
+ PrometheusMetricsService service = new PrometheusMetricsService();
+ ReflectionTestUtils.invokeMethod(service, "register", config);
+ ReflectionTestUtils.invokeMethod(service, "register", config);
+ assertEquals(2.0,
CollectorRegistry.defaultRegistry.getSampleValue(LabelNames.LOGGING_QUEUE_SIZE,
+ new String[]{"collector", "mode"}, new
String[]{getClass().getName(), "single"}));
+ } finally {
+ metrics.unregister();
+ CollectorRegistry.defaultRegistry.clear();
+ }
+ }
+
@Test
public void testRegistered() throws NoSuchFieldException,
IllegalAccessException {
AtomicBoolean registered = (AtomicBoolean)
ReflectUtils.getFieldValue(SERVICE, "registered");
