This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 184cd00f9 fix(cluster,metrics): correct IPv6, PEM, query URI and
dashboard defects (#7009)
184cd00f9 is described below
commit 184cd00f95fe9e556da8a5770e0f8c238f5b561a
Author: Wang1rrr <[email protected]>
AuthorDate: Tue Oct 27 12:05:19 2026 +0800
fix(cluster,metrics): correct IPv6, PEM, query URI and dashboard defects
(#7009)
* fix(metrics): preserve data source URI paths and query parameters (#7009)
The metrics data source address was joined to the query path by string
concatenation after stripping a trailing slash, so a base URL carrying a
path
prefix or a query string produced a wrong endpoint and re-encoded characters
that were already encoded. PrometheusQueryUri now appends path segments
through
URI components and builds the final URI once.
* fix(metrics): encode Basic authentication credentials as UTF-8 (#7011)
Spring's two-argument setBasicAuth encodes as ISO-8859-1: it throws for CJK
or
emoji credentials and silently produces the wrong bytes for accented
Latin-1.
Pass StandardCharsets.UTF_8 explicitly at both call sites.
* fix(nameserver): reject IPv6 network prefixes in endpoints (#6970)
An IPv6 literal carrying a CIDR prefix length parsed as a nameserver
address and
was later handed to the client, which cannot connect to a prefix. Treat a
`/`
inside an IPv6 literal as invalid.
* fix(message): normalize IPv6 broker topology endpoints (#6975)
The known-broker-endpoint guard compared hosts textually, so in an IPv6
cluster
a legitimate message id was rejected as referring to an unknown broker.
Normalize a host that contains a colon through InetAddress before adding it
to
the set - no DNS resolution is introduced.
* fix(instance): cancel queued resource counts on shutdown (#6978)
close() shut the executor down but left the queued futures alone, so
shutdown
blocked for the full 3s await and then reported a state that never happened.
Cancel the futures returned by shutdownNow().
* fix(k8s): parse PEM bundles without joining Base64 blocks (#6989)
parseCertificate stripped the PEM armour, concatenated the Base64 of every
certificate in the bundle and rewrapped it, which corrupts any bundle
holding
more than one certificate. Hand the PEM text to the X.509 reader as-is.
* fix(grafana): query native broker metrics by node ID (#7004)
The native broker metrics only carry a node_id label, so the $broker
variable
and the three native TPS panels matched nothing. Query them by node_id and
regenerate the twelve dashboards from gen_grafana_dashboards.py.
Batch integration of seven related fixes by the same contributor; #6970,
#6975,
#6978, #6989, #7004 and #7011 are closed as merged into this one.
---
server/scripts/gen_grafana_dashboards.py | 10 +--
server/scripts/test_gen_grafana_dashboards.py | 65 +++++++++++++++++-
.../studio/cluster/k8s/K8sCertService.java | 13 ++--
.../AbstractPrometheusCompatibleMetricsSource.java | 8 +--
.../studio/cluster/metrics/PrometheusQueryUri.java | 42 ++++++++++++
.../cluster/nameserver/NamesrvAddrParser.java | 3 +-
.../instance/InstanceResourceCountRunner.java | 6 +-
.../provider/apache/BrokerTopologyGuards.java | 24 ++++++-
.../rocketmq/studio/settings/SettingsService.java | 15 ++--
.../main/resources/grafana/rocketmq-broker.json | 10 +--
.../main/resources/grafana/rocketmq-consumer.json | 2 +-
.../src/main/resources/grafana/rocketmq-dlq.json | 2 +-
.../src/main/resources/grafana/rocketmq-jvm.json | 2 +-
.../main/resources/grafana/rocketmq-latency.json | 2 +-
.../main/resources/grafana/rocketmq-network.json | 2 +-
.../main/resources/grafana/rocketmq-overview.json | 4 +-
.../main/resources/grafana/rocketmq-producer.json | 2 +-
.../main/resources/grafana/rocketmq-storage.json | 2 +-
.../resources/grafana/rocketmq-threadpool.json | 2 +-
.../src/main/resources/grafana/rocketmq-topic.json | 2 +-
.../src/main/resources/grafana/rocketmq-tps.json | 6 +-
.../studio/cluster/k8s/K8sCertServiceTest.java | 67 ++++++++++++++++++
.../metrics/PrometheusMetricsSourceTest.java | 42 ++++++++++++
.../cluster/nameserver/NamesrvAddrParserTest.java | 16 +++++
.../instance/InstanceResourceCountRunnerTest.java | 39 +++++++++++
.../apache/BrokerTopologyGuardsIpv6Test.java | 79 ++++++++++++++++++++++
.../studio/settings/SettingsServiceTest.java | 40 +++++++++++
27 files changed, 457 insertions(+), 50 deletions(-)
diff --git a/server/scripts/gen_grafana_dashboards.py
b/server/scripts/gen_grafana_dashboards.py
index aefc48962..0e2aae190 100644
--- a/server/scripts/gen_grafana_dashboards.py
+++ b/server/scripts/gen_grafana_dashboards.py
@@ -96,7 +96,7 @@ def template_vars(extra=None):
"name": "broker",
"type": "query",
"datasource": {"type": "prometheus", "uid": DS},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0,
@@ -173,7 +173,7 @@ specs.append((
stat_panel(3, "Total Topics", "count(count by (topic)
(rocketmq_messages_in_total{cluster=\"$cluster\"}))", 6, 8),
stat_panel(4, "Total Consumer Groups", "count(count by (group)
(rocketmq_messages_out_total{cluster=\"$cluster\"}))", 6, 8),
stat_panel(5, "Producer Count",
"max(rocketmq_producer_count{cluster=\"$cluster\"})", 6, 8),
- stat_panel(6, "Broker Count", "count(count by (broker)
(rocketmq_messages_in_total{cluster=\"$cluster\"}))", 6, 8),
+ stat_panel(6, "Broker Count", "count(count by (node_id)
(rocketmq_messages_in_total{cluster=\"$cluster\"}))", 6, 8),
],
))
@@ -182,8 +182,8 @@ specs.append((
"rocketmq-broker", "RocketMQ Broker",
"Per-broker throughput, dispatch backlog and thread pool pressure.",
[
- ts_panel(1, "Broker Messages In TPS", "sum by (broker)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",broker=\"$broker\"}[1m]))",
12, 0, "{{broker}}", "ops"),
- ts_panel(2, "Broker Messages Out TPS", "sum by (broker)
(rate(rocketmq_messages_out_total{cluster=\"$cluster\",broker=\"$broker\"}[1m]))",
12, 0, "{{broker}}", "ops"),
+ ts_panel(1, "Broker Messages In TPS", "sum by (node_id)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",node_id=\"$broker\"}[1m]))",
12, 0, "{{node_id}}", "ops"),
+ ts_panel(2, "Broker Messages Out TPS", "sum by (node_id)
(rate(rocketmq_messages_out_total{cluster=\"$cluster\",node_id=\"$broker\"}[1m]))",
12, 0, "{{node_id}}", "ops"),
ts_panel(3, "Dispatch Behind Bytes",
"rocketmq_dispatch_behind_bytes{cluster=\"$cluster\",broker=\"$broker\"}", 12,
8, "{{broker}}", "bytes"),
ts_panel(4, "Thread Pool Queue Size",
"rocketmq_threadpool_queue_size{cluster=\"$cluster\",broker=\"$broker\"}", 12,
8, "{{broker}}"),
],
@@ -229,7 +229,7 @@ specs.append((
[
ts_panel(1, "Cluster TPS In",
"sum(rate(rocketmq_messages_in_total{cluster=\"$cluster\"}[1m]))", 12, 0, "",
"ops"),
ts_panel(2, "Cluster TPS Out",
"sum(rate(rocketmq_messages_out_total{cluster=\"$cluster\"}[1m]))", 12, 0, "",
"ops"),
- ts_panel(3, "Per-Broker TPS In", "sum by (broker)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",broker=\"$broker\"}[1m]))",
24, 8, "{{broker}}", "ops"),
+ ts_panel(3, "Per-Broker TPS In", "sum by (node_id)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",node_id=\"$broker\"}[1m]))",
24, 8, "{{node_id}}", "ops"),
],
))
diff --git a/server/scripts/test_gen_grafana_dashboards.py
b/server/scripts/test_gen_grafana_dashboards.py
index bbb36b400..a50477df5 100644
--- a/server/scripts/test_gen_grafana_dashboards.py
+++ b/server/scripts/test_gen_grafana_dashboards.py
@@ -18,12 +18,15 @@
"""Regression tests for the Grafana dashboard asset generator."""
import json
+import shutil
+import subprocess
import sys
+import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
-from gen_grafana_dashboards import gauge_panel, layout_panels, specs
+from gen_grafana_dashboards import gauge_panel, layout_panels, specs,
template_vars
DASHBOARD_DIR = Path(__file__).resolve().parent.parent / "src" / "main" /
"resources" / "grafana"
@@ -81,12 +84,12 @@ class OverviewPanelsTest(unittest.TestCase):
panels = {item["title"]: item for item in overview[3]}
expr = panels["Broker Count"]["targets"][0]["expr"]
- # rocketmq_messages_in_total carries both broker and topic labels -
the same dashboard
+ # rocketmq_messages_in_total carries both node_id and topic labels -
the same dashboard
# derives its $broker and $topic template variables from them - so a
bare count() returns
# one series per broker/topic pair and reads hundreds on a two-broker
cluster. The
# neighbouring "Total Topics" panel already counts with count(count by
(topic) (...)).
self.assertEqual(
- 'count(count by (broker)
(rocketmq_messages_in_total{cluster="$cluster"}))',
+ 'count(count by (node_id)
(rocketmq_messages_in_total{cluster="$cluster"}))',
expr,
)
@@ -100,6 +103,62 @@ class OverviewPanelsTest(unittest.TestCase):
)
+class NativeBrokerLabelsTest(unittest.TestCase):
+
+ def test_broker_variable_uses_native_node_id_on_every_dashboard(self):
+ expected =
'label_values(rocketmq_messages_in_total{cluster="$cluster"}, node_id)'
+ broker = next(item for item in template_vars()["list"] if item["name"]
== "broker")
+ self.assertEqual(broker["query"], expected)
+ for path in DASHBOARD_DIR.glob("*.json"):
+ with self.subTest(dashboard=path.stem):
+ model = json.loads(path.read_text(encoding="utf-8"))
+ broker = next(item for item in model["templating"]["list"] if
item["name"] == "broker")
+ self.assertEqual(broker["query"], expected)
+
+ def
test_native_broker_panels_match_generator_and_keep_node_id_legends(self):
+ for uid, _, _, panels in specs:
+ for item in panels:
+ expr = item["targets"][0]["expr"]
+ if "rocketmq_messages_" not in expr or "$broker" not in expr:
+ continue
+ with self.subTest(dashboard=uid, panel=item["title"]):
+ self.assertIn('node_id="$broker"', expr)
+ self.assertIn("sum by (node_id)", expr)
+ self.assertEqual(item["targets"][0]["legendFormat"],
"{{node_id}}")
+ self.assertEqual(shipped_panel_expr(uid, item["title"]),
expr)
+
+ @unittest.skipUnless(shutil.which("promtool"), "promtool is required for
PromQL evaluation")
+ def test_native_broker_count_and_tps_against_prometheus(self):
+ # RocketMQ 5 native metrics carry node_id, not the legacy exporter
broker label.
+ series = [
+
('rocketmq_messages_in_total{cluster="prod",node_id="broker-a",topic="orders"}',
"0+15x4"),
+
('rocketmq_messages_in_total{cluster="prod",node_id="broker-a",topic="payments"}',
"0+30x4"),
+
('rocketmq_messages_in_total{cluster="prod",node_id="broker-b",topic="orders"}',
"0+150x4"),
+
('rocketmq_messages_in_total{cluster="other",node_id="broker-c",topic="orders"}',
"0+1500x4"),
+
('rocketmq_messages_out_total{cluster="prod",node_id="broker-a",topic="orders"}',
"0+45x4"),
+
('rocketmq_messages_out_total{cluster="prod",node_id="broker-b",topic="orders"}',
"0+150x4"),
+ ]
+ cases = []
+ for uid, title, labels, value in [
+ ("rocketmq-overview", "Broker Count", "{}", 2),
+ ("rocketmq-broker", "Broker Messages In TPS",
'{node_id="broker-a"}', 3),
+ ("rocketmq-broker", "Broker Messages Out TPS",
'{node_id="broker-a"}', 3),
+ ("rocketmq-tps", "Per-Broker TPS In", '{node_id="broker-a"}', 3),
+ ]:
+ expr = shipped_panel_expr(uid, title).replace("$cluster",
"prod").replace("$broker", "broker-a")
+ cases.append({"expr": expr, "eval_time": "1m",
+ "exp_samples": [{"labels": labels, "value": value}]})
+ fixture = {"evaluation_interval": "15s", "fuzzy_compare": True,
"tests": [{
+ "interval": "15s", "input_series": [{"series": name, "values":
values} for name, values in series],
+ "promql_expr_test": cases,
+ }]}
+ with tempfile.TemporaryDirectory() as directory:
+ path = Path(directory) / "native-broker.json"
+ path.write_text(json.dumps(fixture), encoding="utf-8")
+ result = subprocess.run(["promtool", "test", "rules", str(path)],
capture_output=True, text=True)
+ self.assertEqual(result.returncode, 0, result.stdout +
result.stderr)
+
+
class ShippedDashboardPanelsTest(unittest.TestCase):
def test_dlq_resend_count_panel_counts_over_the_window(self):
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
index 76eee81ee..23cd71b1b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
@@ -25,6 +25,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
import java.security.cert.CertificateFactory;
import java.security.cert.CertificateParsingException;
import java.security.cert.X509Certificate;
@@ -239,14 +240,14 @@ public class K8sCertService {
}
private X509Certificate parseCertificate(String certPem) {
- String base64 = certPem
- .replace("-----BEGIN CERTIFICATE-----", "")
- .replace("-----END CERTIFICATE-----", "")
- .replaceAll("\\s", "");
try {
+ // Let the X.509 reader consume the first PEM block. Joining every
block's Base64
+ // body fails when the leaf's padding appears before an
intermediate certificate.
+ byte[] encoded = certPem.contains("-----BEGIN CERTIFICATE-----")
+ ? certPem.getBytes(StandardCharsets.UTF_8)
+ : Base64.getDecoder().decode(certPem.replaceAll("\\s",
""));
CertificateFactory factory =
CertificateFactory.getInstance("X.509");
- return (X509Certificate) factory.generateCertificate(
- new
ByteArrayInputStream(Base64.getDecoder().decode(base64)));
+ return (X509Certificate) factory.generateCertificate(new
ByteArrayInputStream(encoded));
} catch (Exception exception) {
throw new BusinessException(400, "Invalid certificate content,
expected a PEM encoded X.509 certificate");
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
index b22502cce..4e8bcf396 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
@@ -39,6 +39,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.net.SocketTimeoutException;
import java.net.URI;
+import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -172,14 +173,11 @@ public abstract class
AbstractPrometheusCompatibleMetricsSource implements Metri
}
try {
String baseUrl = settings.getBaseUrl().strip();
- while (baseUrl.endsWith("/")) {
- baseUrl = baseUrl.substring(0, baseUrl.length() - 1);
- }
// SSRF guard: a stored data source is queried server-side on
every request, so the
// host must be validated here even though it was checked when the
data source was
// saved (a pre-save check alone is bypassable via direct DB
edits).
validateQueryHost(baseUrl);
- URI uri = URI.create(baseUrl + settings.getQueryPath());
+ URI uri = PrometheusQueryUri.appendPath(URI.create(baseUrl),
settings.getQueryPath());
if (!"http".equalsIgnoreCase(uri.getScheme()) &&
!"https".equalsIgnoreCase(uri.getScheme())) {
throw new IllegalArgumentException("Unsupported " +
backendLabel() + " URL scheme");
}
@@ -211,7 +209,7 @@ public abstract class
AbstractPrometheusCompatibleMetricsSource implements Metri
throw new
PrometheusException(HttpStatus.SERVICE_UNAVAILABLE.value(),
backendLabel() + " basic authentication is
incomplete");
}
- headers.setBasicAuth(settings.getUsername(),
settings.getPassword());
+ headers.setBasicAuth(settings.getUsername(),
settings.getPassword(), StandardCharsets.UTF_8);
}
case "bearer" -> {
if (!StringUtils.hasText(settings.getBearerToken())) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusQueryUri.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusQueryUri.java
new file mode 100644
index 000000000..49d55bb8e
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusQueryUri.java
@@ -0,0 +1,42 @@
+/*
+ * 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.rocketmq.studio.cluster.metrics;
+
+import org.springframework.web.util.UriComponentsBuilder;
+
+import java.net.URI;
+
+/** Appends a query endpoint without treating an existing query or fragment as
part of the path. */
+public final class PrometheusQueryUri {
+ private PrometheusQueryUri() {
+ }
+
+ public static URI appendPath(URI baseUri, String queryPath) {
+ URI encodedBase = URI.create(baseUri.toASCIIString());
+ String path = encodedBase.getRawPath();
+ if (path == null) {
+ path = "";
+ }
+ while (path.endsWith("/")) {
+ path = path.substring(0, path.length() - 1);
+ }
+ return UriComponentsBuilder.fromUri(encodedBase)
+ .replacePath(path + queryPath)
+ .build(true)
+ .toUri();
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParser.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParser.java
index 9bebcfd4d..281e160c8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParser.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParser.java
@@ -92,6 +92,7 @@ public final class NamesrvAddrParser {
}
private static boolean isValidIpv6Literal(String ipv6) {
- return ADDRESS_VALIDATOR.isValidInet6Address(ipv6);
+ // The validator also accepts CIDR prefixes, which are not socket
endpoints.
+ return ipv6.indexOf('/') < 0 &&
ADDRESS_VALIDATOR.isValidInet6Address(ipv6);
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
index 319b01e58..242821ddd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
@@ -219,7 +219,11 @@ final class InstanceResourceCountRunner implements
AutoCloseable {
@Override
public void close() {
- executor.shutdownNow();
+ for (Runnable queued : executor.shutdownNow()) {
+ if (queued instanceof Future<?> future) {
+ future.cancel(false);
+ }
+ }
executor.purge();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
index 3ca0fca6f..4f131f944 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuards.java
@@ -16,8 +16,10 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
+import java.net.UnknownHostException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
@@ -101,10 +103,30 @@ public final class BrokerTopologyGuards {
}
for (String brokerAddr : brokerData.getBrokerAddrs().values()) {
if (StringUtils.hasText(brokerAddr)) {
- endpoints.add(brokerAddr.trim());
+ endpoints.add(normalizeIpv6Endpoint(brokerAddr.trim()));
}
}
}
return endpoints;
}
+
+ private static String normalizeIpv6Endpoint(String endpoint) {
+ int separator = endpoint.lastIndexOf(':');
+ if (separator <= 0) {
+ return endpoint;
+ }
+ String host = endpoint.substring(0, separator);
+ if (host.startsWith("[") && host.endsWith("]")) {
+ host = host.substring(1, host.length() - 1);
+ }
+ // Only IPv6 literals are parsed: hostnames must not cause a DNS
lookup here.
+ if (!host.contains(":")) {
+ return endpoint;
+ }
+ try {
+ return InetAddress.getByName(host).getHostAddress() +
endpoint.substring(separator);
+ } catch (UnknownHostException ignored) {
+ return endpoint;
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
b/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
index 9c3e8c0b6..8fd16dc17 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
@@ -22,6 +22,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.cluster.metrics.MetricsBackendType;
+import org.apache.rocketmq.studio.cluster.metrics.PrometheusQueryUri;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.util.UrlHostGuard;
@@ -46,6 +47,7 @@ import java.io.IOException;
import java.net.SocketTimeoutException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
@@ -397,7 +399,7 @@ public class SettingsService {
if (!StringUtils.hasText(request.getUsername()) ||
!StringUtils.hasText(request.getPassword())) {
throw new IllegalArgumentException("Basic authentication
requires username and password");
}
- headers.setBasicAuth(request.getUsername().trim(),
request.getPassword());
+ headers.setBasicAuth(request.getUsername().trim(),
request.getPassword(), StandardCharsets.UTF_8);
return;
}
if (AUTH_BEARER.equals(auth)) {
@@ -421,12 +423,7 @@ public class SettingsService {
if (!StringUtils.hasText(baseUrl)) {
throw new IllegalArgumentException("Data source URL is required");
}
- String normalized = baseUrl.strip();
- while (normalized.endsWith("/")) {
- normalized = normalized.substring(0, normalized.length() - 1);
- }
-
- URI baseUri = new URI(normalized);
+ URI baseUri = new URI(baseUrl.strip());
if (!"http".equalsIgnoreCase(baseUri.getScheme()) &&
!"https".equalsIgnoreCase(baseUri.getScheme())) {
throw new IllegalArgumentException("Data source URL must start
with http:// or https://";);
}
@@ -435,9 +432,9 @@ public class SettingsService {
"Data source URL must not point to a local or private
address");
}
String queryPath = backend.getInstantQueryPath();
- return UriComponentsBuilder.fromUriString(normalized + queryPath)
+ return
UriComponentsBuilder.fromUri(PrometheusQueryUri.appendPath(baseUri, queryPath))
.queryParam("query", PROMETHEUS_TEST_QUERY)
- .build()
+ .build(true)
.toUri();
}
diff --git a/server/src/main/resources/grafana/rocketmq-broker.json
b/server/src/main/resources/grafana/rocketmq-broker.json
index 08f7b7e2b..48b44006d 100644
--- a/server/src/main/resources/grafana/rocketmq-broker.json
+++ b/server/src/main/resources/grafana/rocketmq-broker.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
@@ -98,8 +98,8 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "expr": "sum by (broker)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",broker=\"$broker\"}[1m]))",
- "legendFormat": "{{broker}}",
+ "expr": "sum by (node_id)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",node_id=\"$broker\"}[1m]))",
+ "legendFormat": "{{node_id}}",
"refId": "A"
}
]
@@ -140,8 +140,8 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "expr": "sum by (broker)
(rate(rocketmq_messages_out_total{cluster=\"$cluster\",broker=\"$broker\"}[1m]))",
- "legendFormat": "{{broker}}",
+ "expr": "sum by (node_id)
(rate(rocketmq_messages_out_total{cluster=\"$cluster\",node_id=\"$broker\"}[1m]))",
+ "legendFormat": "{{node_id}}",
"refId": "A"
}
]
diff --git a/server/src/main/resources/grafana/rocketmq-consumer.json
b/server/src/main/resources/grafana/rocketmq-consumer.json
index 80f6e9f62..7effe07c9 100644
--- a/server/src/main/resources/grafana/rocketmq-consumer.json
+++ b/server/src/main/resources/grafana/rocketmq-consumer.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-dlq.json
b/server/src/main/resources/grafana/rocketmq-dlq.json
index 35d30d915..48b4e0aa0 100644
--- a/server/src/main/resources/grafana/rocketmq-dlq.json
+++ b/server/src/main/resources/grafana/rocketmq-dlq.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-jvm.json
b/server/src/main/resources/grafana/rocketmq-jvm.json
index c19c9cee1..3e38777ee 100644
--- a/server/src/main/resources/grafana/rocketmq-jvm.json
+++ b/server/src/main/resources/grafana/rocketmq-jvm.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-latency.json
b/server/src/main/resources/grafana/rocketmq-latency.json
index fa7e42fc4..6cc3f13f5 100644
--- a/server/src/main/resources/grafana/rocketmq-latency.json
+++ b/server/src/main/resources/grafana/rocketmq-latency.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-network.json
b/server/src/main/resources/grafana/rocketmq-network.json
index b6085a90e..7f6fc3d75 100644
--- a/server/src/main/resources/grafana/rocketmq-network.json
+++ b/server/src/main/resources/grafana/rocketmq-network.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-overview.json
b/server/src/main/resources/grafana/rocketmq-overview.json
index 15119330d..43fb55758 100644
--- a/server/src/main/resources/grafana/rocketmq-overview.json
+++ b/server/src/main/resources/grafana/rocketmq-overview.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
@@ -313,7 +313,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "expr": "count(count by (broker)
(rocketmq_messages_in_total{cluster=\"$cluster\"}))",
+ "expr": "count(count by (node_id)
(rocketmq_messages_in_total{cluster=\"$cluster\"}))",
"refId": "A"
}
]
diff --git a/server/src/main/resources/grafana/rocketmq-producer.json
b/server/src/main/resources/grafana/rocketmq-producer.json
index c9f06274a..de2e0eba1 100644
--- a/server/src/main/resources/grafana/rocketmq-producer.json
+++ b/server/src/main/resources/grafana/rocketmq-producer.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-storage.json
b/server/src/main/resources/grafana/rocketmq-storage.json
index 4952427ca..aed296d23 100644
--- a/server/src/main/resources/grafana/rocketmq-storage.json
+++ b/server/src/main/resources/grafana/rocketmq-storage.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-threadpool.json
b/server/src/main/resources/grafana/rocketmq-threadpool.json
index c40b0044f..a4ff9a1d3 100644
--- a/server/src/main/resources/grafana/rocketmq-threadpool.json
+++ b/server/src/main/resources/grafana/rocketmq-threadpool.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-topic.json
b/server/src/main/resources/grafana/rocketmq-topic.json
index 94093ab7a..915383275 100644
--- a/server/src/main/resources/grafana/rocketmq-topic.json
+++ b/server/src/main/resources/grafana/rocketmq-topic.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
diff --git a/server/src/main/resources/grafana/rocketmq-tps.json
b/server/src/main/resources/grafana/rocketmq-tps.json
index 062bd6bee..9bba7bdca 100644
--- a/server/src/main/resources/grafana/rocketmq-tps.json
+++ b/server/src/main/resources/grafana/rocketmq-tps.json
@@ -37,7 +37,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, broker)",
+ "query":
"label_values(rocketmq_messages_in_total{cluster=\"$cluster\"}, node_id)",
"refresh": 2,
"current": {},
"hide": 0
@@ -182,8 +182,8 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "expr": "sum by (broker)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",broker=\"$broker\"}[1m]))",
- "legendFormat": "{{broker}}",
+ "expr": "sum by (node_id)
(rate(rocketmq_messages_in_total{cluster=\"$cluster\",node_id=\"$broker\"}[1m]))",
+ "legendFormat": "{{node_id}}",
"refId": "A"
}
]
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
index 123adafbd..86d2188bf 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
@@ -85,6 +85,29 @@ class K8sCertServiceTest {
-----END CERTIFICATE-----
""";
+ // Self-signed fixture with padded Base64: concatenating decoded PEM
bodies is not valid Base64.
+ private static final String PADDED_CERT_PEM = """
+ -----BEGIN CERTIFICATE-----
+ MIIC/TCCAeWgAwIBAgIIGOSWQYsmrhswDQYJKoZIhvcNAQEMBQAwHTEbMBkGA1UE
+ AxMSY2hhaW4tbGVhZi1maXh0dXJlMB4XDTI2MTAxOTAzMzE1NFoXDTI3MTAxOTAz
+ MzE1NFowHTEbMBkGA1UEAxMSY2hhaW4tbGVhZi1maXh0dXJlMIIBIjANBgkqhkiG
+ 9w0BAQEFAAOCAQ8AMIIBCgKCAQEA119FOU0/w31UeWkSQzCCPRZpojS+VF6Ossg7
+ qCODIqjxrSUcaGXbT0oMEOZXSuXyLxFQyPYp2TNzhX03k60kaKrP/Aep7tSg0Evh
+ IbDvI4sudrAwuTPW2gxm7LJewKVhANzebepiXHcEQEw80Y6DL42HEVS2hNMEuQ8B
+ pNTB/Y9/w0pOqdniewuEAujYzYvpalEsAfk9kgyFSo0n0R58QAVC+9zkZG96JthT
+ OI3+MWz75jSlVe3DLCUn6UZAvSDLmfcfbLqZiIiWlIwO8G/sTUd5x4JeA2wV1tIc
+ rqs4wH6jUcn4ntBxj8XlP3hGMv/stKHs7eWgntIQr6V1E29W0QIDAQABo0EwPzAd
+ BgNVHQ4EFgQU9fRXroIZUDtHInqnB0cOm4t9IEMwHgYDVR0RBBcwFYITYnJva2Vy
+ LmV4YW1wbGUudGVzdDANBgkqhkiG9w0BAQwFAAOCAQEAcTRl79KET8Dgks6OrwEn
+ gseClYO/pm8RKsUbJU0fZiw6lASH2LIZEIohcMamaIhj0//AhZUKtKA/W+PaQR/o
+ G8kvr1kkedUl1+6LSRmDJaPiRWkcsrwIsuBnSzlJWWGhJSO88oZfDYK5w8mZiWDY
+ 6VlwWtGWpFmA2nUS/KvySHxN3CHF2wDX6S2+oo85i+eqNTVjKhDS4yCXV+n2BEyC
+ 2g2mHuhkeJUT1jaPD1SEwO09N2Yf0Yz0j9glZhQdShlAPq029jfv9x8PWv2/iaB4
+ VHNEY4xAlzAMvSQ1LckL46dCnZGNE9XLUi4CVRJjZrItfWxG3LfL3ojqWaQX+EyW
+ JQ==
+ -----END CERTIFICATE-----
+ """;
+
@Mock
private K8sCertRepository k8sCertRepository;
@@ -114,6 +137,50 @@ class K8sCertServiceTest {
sampleCert.setGmtModified(LocalDateTime.of(2025, 1, 2, 0, 0));
}
+ @Test
+ void createCertShouldReadFirstCertificateFromPemBundleTest() throws
Exception {
+ when(k8sCertRepository.save(any(K8sCertVO.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+ X509Certificate first = (X509Certificate)
CertificateFactory.getInstance("X.509")
+ .generateCertificate(new
ByteArrayInputStream(PADDED_CERT_PEM.getBytes(StandardCharsets.UTF_8)));
+
+ for (String newline : List.of("\n", "\r\n")) {
+ String bundle = (PADDED_CERT_PEM + TEST_CERT_PEM).replace("\n",
newline);
+ K8sCertVO result =
k8sCertService.createCert(CreateCertDTO.builder()
+ .k8sId("bundle-cert").type("TLS").certPem(bundle).build());
+
+ assertThat(result.getCertPem()).isEqualTo(bundle);
+
assertThat(result.getIssuer()).isEqualTo(first.getIssuerX500Principal().getName());
+ assertThat(result.getSan()).containsExactly("broker.example.test");
+ assertThat(result.getNotAfter())
+
.isEqualTo(LocalDateTime.ofInstant(first.getNotAfter().toInstant(),
ZoneOffset.UTC));
+ }
+ }
+
+ @Test
+ void createCertShouldRetainBareBase64CompatibilityTest() {
+ when(k8sCertRepository.save(any(K8sCertVO.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+ String base64 = PADDED_CERT_PEM.replace("-----BEGIN CERTIFICATE-----",
"")
+ .replace("-----END CERTIFICATE-----", "");
+
+ K8sCertVO result = k8sCertService.createCert(CreateCertDTO.builder()
+ .type("TLS").certPem(base64).build());
+
+ assertThat(result.getIssuer()).isEqualTo("CN=chain-leaf-fixture");
+ assertThat(result.getCertPem()).isEqualTo(base64);
+ }
+
+ @Test
+ void createCertShouldStillRejectMalformedPemTest() {
+ CreateCertDTO command = CreateCertDTO.builder().type("TLS")
+ .certPem("-----BEGIN CERTIFICATE-----\ninvalid\n-----END
CERTIFICATE-----").build();
+
+ assertThatThrownBy(() -> k8sCertService.createCert(command))
+
.isInstanceOf(BusinessException.class).hasMessageContaining("Invalid
certificate content");
+ verifyNoInteractions(k8sCertRepository, operationAuditService);
+ }
+
@Test
void listCertsShouldReturnAllCerts() {
K8sCertVO secondCert = K8sCertVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
index b228393c4..309da0d78 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
@@ -22,6 +22,9 @@ import com.sun.net.httpserver.HttpServer;
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.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.http.HttpStatus;
import org.springframework.web.client.RestClient;
@@ -164,6 +167,23 @@ class PrometheusMetricsSourceTest {
assertThat(series.getHistograms().get(0).getHistogram().path("count").asText()).isEqualTo("2");
}
+ @ParameterizedTest
+ @ValueSource(strings = {"", "?tenant=a%2Fb+%26", "?route=/",
"?tenant=a%2Fb#settings"})
+ void queryShouldAppendEndpointToPathAndPreserveRawQueryTest(String suffix)
{
+ AtomicReference<String> requestUri = new AtomicReference<>();
+ server.createContext("/", exchange -> {
+ requestUri.set(exchange.getRequestURI().toString());
+ respond(exchange, 200, successResponse());
+ });
+ PrometheusProperties properties = properties(Duration.ofSeconds(2));
+ properties.setBaseUrl(baseUrl + "/tenant%20one/" + suffix);
+
+ source(properties).query(query());
+
+ String transmittedSuffix = suffix.split("#", 2)[0];
+
assertThat(requestUri.get()).isEqualTo("/tenant%20one/api/v1/query_range" +
transmittedSuffix);
+ }
+
@Test
void queryShouldApplyBasicAuthentication() {
AtomicReference<String> authorization = new AtomicReference<>();
@@ -181,6 +201,28 @@ class PrometheusMetricsSourceTest {
assertThat(authorization.get()).isEqualTo("Basic " + credentials);
}
+ @ParameterizedTest
+ @CsvSource({"op\u00e9rator, p\u00e4ssword", "\u7528\u6237, \u5bc6\u7801",
"studio, secret\ud83d\udd12"})
+ void queryShouldEncodeBasicCredentialsAsUtf8Test(String username, String
password) {
+ String expectedAuthorization = "Basic " + Base64.getEncoder()
+ .encodeToString((username + ":" +
password).getBytes(StandardCharsets.UTF_8));
+ AtomicReference<String> authorization = new AtomicReference<>();
+ server.createContext("/api/v1/query_range", exchange -> {
+
authorization.set(exchange.getRequestHeaders().getFirst("Authorization"));
+ if (expectedAuthorization.equals(authorization.get())) {
+ respond(exchange, 200, successResponse());
+ } else {
+ respond(exchange, 401, "Unauthorized");
+ }
+ });
+ PrometheusProperties properties = properties(Duration.ofSeconds(2));
+ properties.setUsername(username);
+ properties.setPassword(password);
+
+
assertThat(source(properties).query(query()).getResultType()).isEqualTo("matrix");
+ assertThat(authorization.get()).isEqualTo(expectedAuthorization);
+ }
+
@Test
void queryShouldPreferBearerAuthentication() {
AtomicReference<String> authorization = new AtomicReference<>();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParserTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParserTest.java
index b5180fc39..a669fd5d1 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParserTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NamesrvAddrParserTest.java
@@ -71,6 +71,22 @@ class NamesrvAddrParserTest {
.isEqualTo("[fe80::abcd%12]:9876");
}
+ @Test
+ void rejectsIpv6NetworkPrefixesInSocketEndpointsTest() {
+ for (String endpoint : new String[] {
+ "[2001:db8::1/0]:9876",
+ "[2001:db8::1/64]:9876",
+ "[2001:db8::1/128]:9876",
+ "[fe80::1%eth0/64]:9876",
+ "[fe80::1%12/128]:9876"
+ }) {
+ assertThatThrownBy(() -> NamesrvAddrParser.normalize(endpoint))
+ .as("network prefix must not be accepted as an endpoint:
%s", endpoint)
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("malformed IPv6");
+ }
+ }
+
@Test
void rejectsBlankInputTest() {
assertThatThrownBy(() -> NamesrvAddrParser.normalize(" "))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
index 4fab0ebc5..b0f7a2874 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
@@ -21,6 +21,7 @@ import org.junit.jupiter.api.Test;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -216,6 +217,44 @@ class InstanceResourceCountRunnerTest {
}
}
+ @Test
+ void closeShouldCancelQueuedCountsAndReleaseWaitingRequestsTest() throws
Exception {
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicBoolean queuedExecuted = new AtomicBoolean();
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(1, 1, 1,
TimeUnit.MINUTES)) {
+ FutureTask<List<InstanceResourceCountRunner.CountOutcome>> call =
new FutureTask<>(
+ () -> runner.countAll(List.of(instance(1L), instance(2L)),
instance -> {
+ if (instance.getId().equals(2L)) {
+ queuedExecuted.set(true);
+ }
+ started.countDown();
+ release.await();
+ return new
InstanceResourceCountRunner.ResourceCounts(1, 1);
+ }));
+ Thread caller = new Thread(call, "resource-count-shutdown-test");
+ caller.start();
+ try {
+ assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(waitForQueueSize(runner, 1, 5,
TimeUnit.SECONDS)).isTrue();
+
+ runner.close();
+
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
call.get(2, TimeUnit.SECONDS);
+ assertThat(outcomes).hasSize(2);
+ assertThat(outcomes.get(1).status())
+
.isEqualTo(InstanceResourceCountRunner.OutcomeStatus.INTERRUPTED);
+ assertThat(queuedExecuted).isFalse();
+ assertThat(runner.queuedCount()).isZero();
+ } finally {
+ release.countDown();
+ call.cancel(true);
+ caller.join(TimeUnit.SECONDS.toMillis(5));
+ }
+ }
+ }
+
private static InstanceVO instance(long id) {
InstanceVO instance = InstanceVO.builder().name("instance-" +
id).build();
instance.setId(id);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuardsIpv6Test.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuardsIpv6Test.java
new file mode 100644
index 000000000..f93027f76
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/BrokerTopologyGuardsIpv6Test.java
@@ -0,0 +1,79 @@
+/*
+ * 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.rocketmq.studio.provider.apache;
+
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class BrokerTopologyGuardsIpv6Test {
+
+ @ParameterizedTest
+ @ValueSource(strings = {"2001:db8::1:10911", "[2001:db8::1]:10911",
+ "[2001:0DB8:0000:0000:0000:0000:0000:0001]:10911",
"2001:db8:0:0:0:0:0:1:10911"})
+ void equivalentIpv6BrokerEndpointsPermitOffsetLookupTest(String
advertised) throws Exception {
+ String messageId = messageId("2001:db8::1");
+ MQAdminExt admin = adminWithEndpoint(advertised);
+
+ assertThat(BrokerTopologyGuards.isWithinKnownBrokerTopology(admin,
messageId)).isTrue();
+ assertThat(BrokerTopologyGuards.validatedBrokerAddr(
+ admin, messageId, MessageDecoder.decodeMessageId(messageId)))
+ .isEqualTo("2001:db8:0:0:0:0:0:1:10911");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"[2001:db8::2]:10911", "[2001:db8::1]:10912",
+ "[2001:db8::1/64]:10911", "[not:an:ipv6]:10911",
"broker.example.invalid:10911"})
+ void differentOrInvalidIpv6EndpointsRemainRejectedTest(String advertised)
throws Exception {
+ assertThat(BrokerTopologyGuards.isWithinKnownBrokerTopology(
+ adminWithEndpoint(advertised),
messageId("2001:db8::1"))).isFalse();
+ }
+
+ @Test
+ void ipv4EndpointMatchingIsPreservedTest() throws Exception {
+ assertThat(BrokerTopologyGuards.isWithinKnownBrokerTopology(
+ adminWithEndpoint("192.0.2.1:10911"),
messageId("192.0.2.1"))).isTrue();
+ assertThat(BrokerTopologyGuards.isWithinKnownBrokerTopology(
+ adminWithEndpoint("192.0.2.2:10911"),
messageId("192.0.2.1"))).isFalse();
+ }
+
+ private static String messageId(String address) {
+ return MessageDecoder.createMessageId(new InetSocketAddress(address,
10911), 12345L);
+ }
+
+ private static MQAdminExt adminWithEndpoint(String endpoint) throws
Exception {
+ BrokerData broker = new BrokerData();
+ broker.setBrokerAddrs(new HashMap<>(Map.of(0L, endpoint)));
+ ClusterInfo cluster = new ClusterInfo();
+ cluster.setBrokerAddrTable(new HashMap<>(Map.of("broker-a", broker)));
+ MQAdminExt admin = mock(MQAdminExt.class);
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster);
+ return admin;
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
index dd4f8163a..50bc94581 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
@@ -26,6 +26,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -574,6 +575,25 @@ class SettingsServiceTest {
.isEqualTo(400);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"", "?tenant=a%2Fb+%26", "?route=/",
"?tenant=a%2Fb#settings"})
+ void connectionShouldPreserveEncodedBasePathAndQueryTest(String suffix) {
+ int fragmentIndex = suffix.indexOf('#');
+ String query = fragmentIndex < 0 ? suffix : suffix.substring(0,
fragmentIndex);
+ String fragment = fragmentIndex < 0 ? "" :
suffix.substring(fragmentIndex);
+ String expectedUrl = PROMETHEUS_BASE_URL + "/tenant%20one/api/v1/query"
+ + (query.isEmpty() ? "?" : query + "&") + "query=up" +
fragment;
+ prometheusServer.expect(requestTo(expectedUrl))
+ .andExpect(method(HttpMethod.GET))
+ .andRespond(withSuccess(PROMETHEUS_SUCCESS_BODY,
MediaType.APPLICATION_JSON));
+ DataSourceTestDTO request = DataSourceTestDTO.builder()
+ .url(PROMETHEUS_BASE_URL + "/tenant%20one/" + suffix)
+ .type("Prometheus")
+ .build();
+
+
assertThat(settingsService.testDataSource(request).isSuccess()).isTrue();
+ }
+
@Test
void connectionShouldQueryPrometheusEndpointTest() {
prometheusServer.expect(requestTo(PROMETHEUS_QUERY_URL))
@@ -646,6 +666,26 @@ class SettingsServiceTest {
assertThat(result.isSuccess()).isTrue();
}
+ @ParameterizedTest
+ @CsvSource({"op\u00e9rator, p\u00e4ssword", "\u7528\u6237, \u5bc6\u7801",
"studio, secret\ud83d\udd12"})
+ void connectionShouldEncodeBasicCredentialsAsUtf8Test(String username,
String password) {
+ String expectedAuthorization = "Basic " + Base64.getEncoder()
+ .encodeToString((username + ":" +
password).getBytes(StandardCharsets.UTF_8));
+ prometheusServer.expect(requestTo(PROMETHEUS_QUERY_URL))
+ .andExpect(method(HttpMethod.GET))
+ .andExpect(header(HttpHeaders.AUTHORIZATION,
expectedAuthorization))
+ .andRespond(withSuccess(PROMETHEUS_SUCCESS_BODY,
MediaType.APPLICATION_JSON));
+ DataSourceTestDTO request = DataSourceTestDTO.builder()
+ .url(PROMETHEUS_BASE_URL)
+ .type("Prometheus")
+ .auth("Basic Auth")
+ .username(username)
+ .password(password)
+ .build();
+
+
assertThat(settingsService.testDataSource(request).isSuccess()).isTrue();
+ }
+
@Test
void
connectionShouldNormalizeIdentifiersIndependentlyOfDefaultLocaleTest() {
String expectedAuthorization = "Basic "
