This is an automated email from the ASF dual-hosted git repository.
luoyuxia 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 6238daaed [metrics][docs] Support Basic Auth for Prometheus
PushGateway reporter (#3552)
6238daaed is described below
commit 6238daaed660b02b793924569205e50677d378fc
Author: GuoYu <[email protected]>
AuthorDate: Fri Jul 3 16:53:08 2026 +0800
[metrics][docs] Support Basic Auth for Prometheus PushGateway reporter
(#3552)
---
.../org/apache/fluss/config/ConfigOptions.java | 18 ++
.../apache/fluss/config/ConfigurationUtils.java | 4 +
.../prometheus/PrometheusPushGatewayReporter.java | 28 ++-
.../PrometheusPushGatewayReporterPlugin.java | 20 ++-
.../PrometheusPushGatewayReporterTest.java | 190 +++++++++++++++++++++
website/docs/maintenance/configuration.md | 2 +
.../maintenance/observability/metric-reporters.md | 4 +
7 files changed, 262 insertions(+), 4 deletions(-)
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index 0301d3d47..2d97ebab5 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -2171,6 +2171,24 @@ public class ConfigOptions {
.withDescription(
"The interval of pushing metrics to
Prometheus PushGateway.");
+ public static final ConfigOption<String>
METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_USERNAME =
+ key("metrics.reporter.prometheus-push.username")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The username for Basic Auth of the Prometheus
PushGateway. "
+ + "Leave it unset to disable
authentication.");
+
+ public static final ConfigOption<Password>
METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_PASSWORD =
+ key("metrics.reporter.prometheus-push.password")
+ .passwordType()
+ .noDefaultValue()
+ .withDescription(
+ "The password for Basic Auth of the Prometheus
PushGateway. "
+ + "Only takes effect when username is
configured. "
+ + "The value is automatically redacted
when the configuration "
+ + "is logged or displayed.");
+
// ------------------------------------------------------------------------
// ConfigOptions for jmx reporter
// ------------------------------------------------------------------------
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java
b/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java
index 4b4ae09fc..1d0810256 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigurationUtils.java
@@ -99,6 +99,10 @@ public class ConfigurationUtils {
} else if (String.class.equals(clazz)) {
return (T) convertToString(rawValue);
} else if (Password.class.equals(clazz)) {
+ if (rawValue instanceof Password) {
+ return (T) rawValue;
+ }
+
return (T) new Password(convertToString(rawValue));
} else if (clazz.isEnum()) {
return (T) convertToEnum(rawValue, (Class<? extends Enum<?>>)
clazz);
diff --git
a/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporter.java
b/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporter.java
index f6102f9cd..cfe0b858e 100644
---
a/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporter.java
+++
b/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporter.java
@@ -19,14 +19,21 @@ package org.apache.fluss.metrics.prometheus;
import org.apache.fluss.metrics.Metric;
import org.apache.fluss.metrics.reporter.ScheduledMetricReporter;
+import org.apache.fluss.utils.StringUtils;
+import io.prometheus.client.exporter.HttpConnectionFactory;
import io.prometheus.client.exporter.PushGateway;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
+import java.net.HttpURLConnection;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.util.Base64;
import java.util.Map;
/** {@link ScheduledMetricReporter} that pushes {@link Metric Metrics} to
Prometheus PushGateway. */
@@ -46,12 +53,18 @@ public class PrometheusPushGatewayReporter extends
AbstractPrometheusReporter
String jobName,
Map<String, String> groupingKey,
final boolean deleteOnShutdown,
- Duration pushInterval) {
+ Duration pushInterval,
+ @Nullable String username,
+ @Nullable String password) {
this.pushGateway = new PushGateway(hostUrl);
this.jobName = jobName;
this.groupingKey = groupingKey;
this.deleteOnShutdown = deleteOnShutdown;
this.pushInterval = pushInterval;
+ if (!StringUtils.isNullOrWhitespaceOnly(username)) {
+ this.pushGateway.setConnectionFactory(
+ basicAuthConnectionFactory(username, password == null ? ""
: password));
+ }
}
@Override
@@ -80,4 +93,17 @@ public class PrometheusPushGatewayReporter extends
AbstractPrometheusReporter
LOG.warn("Could not push metrics to PushGateway.", e);
}
}
+
+ private static HttpConnectionFactory basicAuthConnectionFactory(String
user, String password) {
+ final String header =
+ "Basic "
+ + Base64.getEncoder()
+ .encodeToString(
+ (user + ":" +
password).getBytes(StandardCharsets.UTF_8));
+ return url -> {
+ HttpURLConnection connection = (HttpURLConnection) new
URL(url).openConnection();
+ connection.setRequestProperty("Authorization", header);
+ return connection;
+ };
+ }
}
diff --git
a/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterPlugin.java
b/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterPlugin.java
index 5786a002c..a1b565109 100644
---
a/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterPlugin.java
+++
b/fluss-metrics/fluss-metrics-prometheus/src/main/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterPlugin.java
@@ -19,6 +19,7 @@ package org.apache.fluss.metrics.prometheus;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.Password;
import org.apache.fluss.metrics.reporter.MetricReporter;
import org.apache.fluss.metrics.reporter.MetricReporterPlugin;
import org.apache.fluss.utils.StringUtils;
@@ -37,8 +38,10 @@ import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_
import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_GROUPING_KEY;
import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_HOST_URL;
import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_JOB_NAME;
+import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_PASSWORD;
import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_PUSH_INTERVAL;
import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_RANDOM_JOB_NAME_SUFFIX;
+import static
org.apache.fluss.config.ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_USERNAME;
/** {@link MetricReporterPlugin} for {@link PrometheusPushGatewayReporter}. */
public class PrometheusPushGatewayReporterPlugin implements
MetricReporterPlugin {
@@ -56,23 +59,34 @@ public class PrometheusPushGatewayReporterPlugin implements
MetricReporterPlugin
boolean randomSuffix =
config.get(METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_RANDOM_JOB_NAME_SUFFIX);
Duration pushInterval =
config.get(METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_PUSH_INTERVAL);
+ String username =
config.get(METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_USERNAME);
+ Password passwordOption =
config.get(METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_PASSWORD);
+ String password = passwordOption == null ? null :
passwordOption.value();
String jobName = configuredJobName;
if (randomSuffix) {
jobName = configuredJobName + new Random().nextLong();
}
Map<String, String> groupingKey =
parseGroupingKey(config.get(METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_GROUPING_KEY));
+ boolean basicAuthEnabled =
!StringUtils.isNullOrWhitespaceOnly(username);
LOG.info(
- "Configured PrometheusPushGatewayReporter with {hostUrl:{},
jobName:{}, randomJobNameSuffix:{}, deleteOnShutdown:{}, groupingKey:{},
pushInterval:{}}",
+ "Configured PrometheusPushGatewayReporter with {hostUrl:{},
jobName:{}, randomJobNameSuffix:{}, deleteOnShutdown:{}, groupingKey:{},
pushInterval:{}, basicAuthEnabled:{}}",
hostUrl,
jobName,
randomSuffix,
deleteOnShutdown,
groupingKey,
- pushInterval);
+ pushInterval,
+ basicAuthEnabled);
try {
return new PrometheusPushGatewayReporter(
- new URL(hostUrl), jobName, groupingKey, deleteOnShutdown,
pushInterval);
+ new URL(hostUrl),
+ jobName,
+ groupingKey,
+ deleteOnShutdown,
+ pushInterval,
+ basicAuthEnabled ? username : null,
+ basicAuthEnabled ? password : null);
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git
a/fluss-metrics/fluss-metrics-prometheus/src/test/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterTest.java
b/fluss-metrics/fluss-metrics-prometheus/src/test/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterTest.java
new file mode 100644
index 000000000..a8d2f6d46
--- /dev/null
+++
b/fluss-metrics/fluss-metrics-prometheus/src/test/java/org/apache/fluss/metrics/prometheus/PrometheusPushGatewayReporterTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.fluss.metrics.prometheus;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.Password;
+import org.apache.fluss.metrics.reporter.MetricReporter;
+
+import com.sun.net.httpserver.HttpExchange;
+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 java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PrometheusPushGatewayReporterTest {
+
+ private HttpServer server;
+ private BlockingQueue<String> receivedAuthHeaders;
+
+ @BeforeEach
+ void startFakePushGateway() throws IOException {
+ receivedAuthHeaders = new ArrayBlockingQueue<>(8);
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext(
+ "/",
+ (HttpExchange exchange) -> {
+ // capture (possibly null) Authorization header, using
empty string as absent
+ String auth =
exchange.getRequestHeaders().getFirst("Authorization");
+ receivedAuthHeaders.offer(auth == null ? "" : auth);
+ // drain request body so client does not block (JDK 8
compatible)
+ try (InputStream body = exchange.getRequestBody()) {
+ byte[] buf = new byte[1024];
+ while (body.read(buf) != -1) {
+ // discard
+ }
+ }
+
+ exchange.sendResponseHeaders(202, -1);
+ exchange.close();
+ });
+ server.start();
+ }
+
+ @AfterEach
+ void stopFakePushGateway() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void reportSendsAuthorizationHeaderWhenBasicAuthConfigured() throws
Exception {
+ PrometheusPushGatewayReporter reporter =
+ new PrometheusPushGatewayReporter(
+ pushGatewayUrl(),
+ "test-job",
+ Collections.emptyMap(),
+ false,
+ Duration.ofSeconds(10),
+ "myuser",
+ "mypassword");
+ try {
+ reporter.report();
+
+ String header = receivedAuthHeaders.poll(5, TimeUnit.SECONDS);
+ assertThat(header).isNotNull().startsWith("Basic ");
+
+ String decoded =
+ new String(
+ Base64.getDecoder().decode(header.substring("Basic
".length())),
+ StandardCharsets.UTF_8);
+ assertThat(decoded).isEqualTo("myuser:mypassword");
+ } finally {
+ reporter.close();
+ }
+ }
+
+ @Test
+ void reportSendsNoAuthorizationHeaderWhenBasicAuthNotConfigured() throws
Exception {
+ PrometheusPushGatewayReporter reporter =
+ new PrometheusPushGatewayReporter(
+ pushGatewayUrl(),
+ "test-job",
+ Collections.emptyMap(),
+ false,
+ Duration.ofSeconds(10),
+ null,
+ null);
+ try {
+ reporter.report();
+
+ String header = receivedAuthHeaders.poll(5, TimeUnit.SECONDS);
+ assertThat(header).isNotNull().isEmpty();
+ } finally {
+ reporter.close();
+ }
+ }
+
+ @Test
+ void reportSendsNoAuthorizationHeaderWhenUsernameIsBlank() throws
Exception {
+ // password without username should NOT enable basic auth
+ PrometheusPushGatewayReporter reporter =
+ new PrometheusPushGatewayReporter(
+ pushGatewayUrl(),
+ "test-job",
+ Collections.emptyMap(),
+ false,
+ Duration.ofSeconds(10),
+ "",
+ "somePwd");
+ try {
+ reporter.report();
+
+ String header = receivedAuthHeaders.poll(5, TimeUnit.SECONDS);
+ assertThat(header).isNotNull().isEmpty();
+ } finally {
+ reporter.close();
+ }
+ }
+
+ @Test
+ void pluginCreatesReporterCarryingBasicAuth() throws Exception {
+ Configuration config = new Configuration();
+ config.setString(
+ ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_HOST_URL,
+ pushGatewayUrl().toString());
+ config.setString(
+
ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_JOB_NAME, "plugin-job");
+ config.setString(
+
ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_USERNAME, "plugUser");
+ config.set(
+ ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_PASSWORD,
+ new Password("plugPwd"));
+ config.setString(
+
ConfigOptions.METRICS_REPORTER_PROMETHEUS_PUSHGATEWAY_GROUPING_KEY, "k1=v1");
+
+ PrometheusPushGatewayReporterPlugin plugin = new
PrometheusPushGatewayReporterPlugin();
+ assertThat(plugin.identifier()).isEqualTo("prometheus-push");
+
+ MetricReporter reporter = plugin.createMetricReporter(config);
+ assertThat(reporter).isInstanceOf(PrometheusPushGatewayReporter.class);
+ try {
+ ((PrometheusPushGatewayReporter) reporter).report();
+
+ String header = receivedAuthHeaders.poll(5, TimeUnit.SECONDS);
+ assertThat(header).isNotNull().startsWith("Basic ");
+ String decoded =
+ new String(
+ Base64.getDecoder().decode(header.substring("Basic
".length())),
+ StandardCharsets.UTF_8);
+ assertThat(decoded).isEqualTo("plugUser:plugPwd");
+ } finally {
+ reporter.close();
+ }
+ }
+
+ private URL pushGatewayUrl() throws IOException {
+ return new URL("http://127.0.0.1:"; + server.getAddress().getPort());
+ }
+}
diff --git a/website/docs/maintenance/configuration.md
b/website/docs/maintenance/configuration.md
index 0a4223074..39e38a4c8 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -198,6 +198,8 @@ More metrics example could be found in [Observability -
Metric Reporters](observ
| metrics.reporter.prometheus-push.random-job-name-suffix | Boolean | true
| Specifies whether a random suffix should be appended to the job name,
defaults to true. This is useful when multiple instances of the reporter are
running on the same host.
|
| metrics.reporter.prometheus-push.delete-on-shutdown | Boolean | true
| Specifies whether to delete metrics from the PushGateway on shutdown,
defaults to true. Fluss will try its best to delete the metrics but this is not
guaranteed.
|
| metrics.reporter.prometheus-push.grouping-key | String | (None)
| Specifies the grouping key which is the group and global labels of all
metrics. The label name and value are separated by '=', and labels are
separated by ';', e.g., k1=v1;k2=v2.
|
+| metrics.reporter.prometheus-push.username | String | (None)
| The username for Basic Auth of the Prometheus PushGateway. Leave it unset
to disable authentication.
|
+| metrics.reporter.prometheus-push.password | String | (None)
| The password for Basic Auth of the Prometheus PushGateway. Only takes
effect when username is configured.
|
## Lakehouse
| Option | Type | Default | Description
|
diff --git a/website/docs/maintenance/observability/metric-reporters.md
b/website/docs/maintenance/observability/metric-reporters.md
index 4afdca5b4..bea71c010 100644
--- a/website/docs/maintenance/observability/metric-reporters.md
+++ b/website/docs/maintenance/observability/metric-reporters.md
@@ -93,6 +93,8 @@ Parameters:
- `metrics.reporter.prometheus-push.random-job-name-suffix` - (Optional)
Specifies whether a random suffix should be appended to the job name, defaults
to true. This is useful when multiple instances of the reporter are running on
the same host.
- `metrics.reporter.prometheus-push.delete-on-shutdown` - (Optional) Specifies
whether to delete metrics from the PushGateway on shutdown, defaults to true.
Fluss will try its best to delete the metrics but this is not guaranteed.
- `metrics.reporter.prometheus-push.grouping-key` - Specifies the grouping key
which is the group and global labels of all metrics. The label name and value
are separated by `=`, and labels are separated by `;`, e.g., `k1=v1;k2=v2`.
+- `metrics.reporter.prometheus-push.username` - (Optional) The username for
Basic Auth of the Prometheus PushGateway. Leave it unset to disable
authentication.
+- `metrics.reporter.prometheus-push.password` - (Optional) The password for
Basic Auth of the Prometheus PushGateway. Only takes effect when `username` is
configured.
Example configuration:
@@ -104,6 +106,8 @@ metrics.reporter.prometheus-push.push-interval: 10 SECONDS
metrics.reporter.prometheus-push.random-job-name-suffix: true
metrics.reporter.prometheus-push.delete-on-shutdown: true
metrics.reporter.prometheus-push.grouping-key:
instance=instance01;cluster=clusterA
+metrics.reporter.prometheus-push.username: myuser
+metrics.reporter.prometheus-push.password: mypassword
```
### InfluxDB
