This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new ba1ce090e [server] Add request queue size metric per processor (#3232)
ba1ce090e is described below
commit ba1ce090e11f5dfc5e685dc3d8fd242e382163b6
Author: Junfan Zhang <[email protected]>
AuthorDate: Fri May 29 08:47:26 2026 +0800
[server] Add request queue size metric per processor (#3232)
Co-authored-by: zhangjunfan <[email protected]>
---
.../fluss/rpc/netty/server/RequestProcessorPool.java | 12 ++++++++++--
.../apache/fluss/rpc/netty/server/RequestsMetrics.java | 5 +++++
.../docs/maintenance/observability/monitor-metrics.md | 18 +++++++++++++++---
3 files changed, 30 insertions(+), 5 deletions(-)
diff --git
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestProcessorPool.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestProcessorPool.java
index 463ebf1ec..a99738003 100644
---
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestProcessorPool.java
+++
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestProcessorPool.java
@@ -38,6 +38,8 @@ import java.util.concurrent.Executors;
final class RequestProcessorPool {
private static final Logger LOG =
LoggerFactory.getLogger(RequestProcessorPool.class);
+ private static final String PROCESSOR_INDEX = "processor_index";
+
private final RequestChannel[] requestChannels;
private final RequestProcessor[] processors;
@@ -54,10 +56,16 @@ final class RequestProcessorPool {
RequestHandler<?>[] requestHandlers =
initializeRequestHandlers(protocols, service);
for (int i = 0; i < numProcessors; i++) {
- requestChannels[i] = new RequestChannel(totalQueueCapacity /
numProcessors);
+ RequestChannel requestChannel = new
RequestChannel(totalQueueCapacity / numProcessors);
+ requestChannels[i] = requestChannel;
// bind processor to a single channel to make requests from the
// same channel processed serializable
- processors[i] = new RequestProcessor(i, requestChannels[i],
service, requestHandlers);
+ processors[i] = new RequestProcessor(i, requestChannel, service,
requestHandlers);
+ requestsMetrics.gauge(
+ PROCESSOR_INDEX,
+ String.valueOf(i),
+ MetricNames.REQUEST_QUEUE_SIZE,
+ requestChannel::requestsCount);
}
// register requestQueueSize metrics
requestsMetrics.gauge(MetricNames.REQUEST_QUEUE_SIZE,
this::getRequestQueueSize);
diff --git
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
index 5bf9ab0d7..592eaf30d 100644
---
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
+++
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
@@ -85,6 +85,11 @@ public class RequestsMetrics {
requestMetricGroup.gauge(name, gauge);
}
+ /** Create a gauge metric in a child group of the request metric group. */
+ <T, G extends Gauge<T>> void gauge(String groupKey, String groupValue,
String name, G gauge) {
+ requestMetricGroup.addGroup(groupKey, groupValue).gauge(name, gauge);
+ }
+
/** Add a metric group for given request name. */
private void addMetrics(MetricGroup parentMetricGroup, String requestName)
{
metricsByRequest.put(
diff --git a/website/docs/maintenance/observability/monitor-metrics.md
b/website/docs/maintenance/observability/monitor-metrics.md
index 3fe38b85f..4fcc83687 100644
--- a/website/docs/maintenance/observability/monitor-metrics.md
+++ b/website/docs/maintenance/observability/monitor-metrics.md
@@ -639,19 +639,31 @@ Some metrics might not be exposed when using other JVM
implementations (e.g. IBM
</thead>
<tbody>
<tr>
- <th rowspan="1"><strong>coordinator</strong></th>
+ <th rowspan="2"><strong>coordinator</strong></th>
<td rowspan="1">request</td>
<td>requestQueueSize</td>
<td>The CoordinatorServer node network waiting queue size.</td>
<td>Gauge</td>
</tr>
<tr>
- <th rowspan="8">tabletserver</th>
+ <td rowspan="1">request_processor_index</td>
+ <td>requestQueueSize</td>
+ <td>The CoordinatorServer node network waiting queue size labeled with
<code>processor_index</code>.</td>
+ <td>Gauge</td>
+ </tr>
+ <tr>
+ <th rowspan="9">tabletserver</th>
<td rowspan="1">request</td>
<td>requestQueueSize</td>
<td>The TabletServer node network waiting queue size.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td rowspan="1">request_processor_index</td>
+ <td>requestQueueSize</td>
+ <td>The TabletServer node network waiting queue size labeled with
<code>processor_index</code>.</td>
+ <td>Gauge</td>
+ </tr>
<tr>
<td rowspan="7">
request_produceLog
@@ -1182,4 +1194,4 @@ All metrics are registered under the
`fluss.tieringService` metric group, which
<td>Meter</td>
</tr>
</tbody>
-</table>
\ No newline at end of file
+</table>
