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 0301d3d47d..2d97ebab56 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 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 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 ad800aa8f6..9c408701c6 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 @@ -67,6 +67,10 @@ public static T convertValue(Object rawValue, Class clazz) { } 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>) 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 f6102f9cdb..cfe0b858e4 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 @@ 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 PrometheusPushGatewayReporter( String jobName, Map 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 void report() { 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 5786a002cd..a1b5651091 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 @@ 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_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 MetricReporter createMetricReporter(Configuration config) { 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 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 0000000000..a8d2f6d46c --- /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 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 0a4223074d..39e38a4c80 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 4afdca5b44..bea71c010d 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