[metric] Support influxdb reporter

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1990,6 +1990,41 @@ public class ConfigOptions {
                                     + "the CoordinatorServer) it is advisable to use a port range "
                                     + "like 9990-9999.");
 
+    // ------------------------------------------------------------------------
+    //  ConfigOptions for influxdb reporter
+    // ------------------------------------------------------------------------
+    public static final ConfigOption<String> METRICS_REPORTER_INFLUXDB_HOST_URL =
+            key("metrics.reporter.influxdb.host-url")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The InfluxDB server host URL including scheme, host name, and port.");
+
+    public static final ConfigOption<String> METRICS_REPORTER_INFLUXDB_BUCKET =
+            key("metrics.reporter.influxdb.bucket")
+                    .stringType()
+                    .noDefaultValue()
+                    .withFallbackKeys("metrics.reporter.influxdb.database")
+                    .withDescription("The InfluxDB bucket/database name.");
+
+    public static final ConfigOption<String> METRICS_REPORTER_INFLUXDB_ORG =
+            key("metrics.reporter.influxdb.org")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The InfluxDB organization name.");
+
+    public static final ConfigOption<String> METRICS_REPORTER_INFLUXDB_TOKEN =
+            key("metrics.reporter.influxdb.token")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The InfluxDB authentication token.");
+
+    public static final ConfigOption<Duration> METRICS_REPORTER_INFLUXDB_PUSH_INTERVAL =
+            key("metrics.reporter.influxdb.push-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(10))
+                    .withDescription("The interval of reporting metrics to InfluxDB.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for lakehouse storage
     // ------------------------------------------------------------------------

fluss-metrics/fluss-metrics-influxdb/pom.xml

@@ -0,0 +1,79 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.fluss</groupId>
+        <artifactId>fluss-metrics</artifactId>
+        <version>0.10-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>fluss-metrics-influxdb</artifactId>
+    <name>Fluss : Metrics : InfluxDB</name>
+
+    <properties>
+        <influxdb3.version>1.8.0</influxdb3.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-common</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.influxdb</groupId>
+            <artifactId>influxdb3-java</artifactId>
+            <version>${influxdb3.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- test dependencies -->
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-test-utils</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-common</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <!-- compilation of main sources -->
+                    <skipMain>${skip.on.java8}</skipMain>
+                    <!-- compilation of test sources -->
+                    <skip>${skip.on.java8}</skip>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

fluss-metrics/fluss-metrics-influxdb/src/main/java/org/apache/fluss/metrics/influxdb/InfluxdbPointProducer.java

@@ -0,0 +1,104 @@
+/*
+ * 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.influxdb;
+
+import org.apache.fluss.metrics.Counter;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.Histogram;
+import org.apache.fluss.metrics.HistogramStatistics;
+import org.apache.fluss.metrics.Meter;
+import org.apache.fluss.metrics.Metric;
+
+import com.influxdb.v3.client.Point;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+/** Producer that creates InfluxDB {@link Point Points} from Fluss {@link Metric Metrics}. */
+public class InfluxdbPointProducer {
+
+    private static final InfluxdbPointProducer INSTANCE = new InfluxdbPointProducer();
+
+    public static InfluxdbPointProducer getInstance() {
+        return INSTANCE;
+    }
+
+    public Point createPoint(
+            Metric metric, String metricName, List<Map.Entry<String, String>> tags, Instant time) {
+        Point point = Point.measurement(metricName).setTimestamp(time);
+
+        for (Map.Entry<String, String> tag : tags) {
+            point.setTag(tag.getKey(), tag.getValue());
+        }
+
+        if (metric instanceof Counter) {
+            return createPointForCounter((Counter) metric, point);
+        }
+
+        if (metric instanceof Gauge) {
+            return createPointForGauge((Gauge<?>) metric, point);
+        }
+
+        if (metric instanceof Meter) {
+            return createPointForMeter((Meter) metric, point);
+        }
+
+        if (metric instanceof Histogram) {
+            return createPointForHistogram((Histogram) metric, point);
+        }
+
+        throw new IllegalArgumentException("Unknown metric type: " + metric.getClass());
+    }
+
+    private Point createPointForCounter(Counter counter, Point point) {
+        return point.setField("count", counter.getCount());
+    }
+
+    private Point createPointForGauge(Gauge<?> gauge, Point point) {
+        Object value = gauge.getValue();
+
+        if (value instanceof Number) {
+            return point.setField("value", ((Number) value));
+        } else if (value instanceof Boolean) {
+            return point.setField("value", ((boolean) value));
+        } else {
+            return point.setField("value", String.valueOf(value));
+        }
+    }
+
+    private Point createPointForMeter(Meter meter, Point point) {
+        return point.setField("rate", meter.getRate()).setField("count", meter.getCount());
+    }
+
+    private Point createPointForHistogram(Histogram histogram, Point point) {
+        HistogramStatistics stats = histogram.getStatistics();
+        return point.setField("count", histogram.getCount())
+                .setField("mean", stats.getMean())
+                .setField("stddev", stats.getStdDev())
+                .setField("min", stats.getMin())
+                .setField("max", stats.getMax())
+                .setField("p50", stats.getQuantile(0.5))
+                .setField("p75", stats.getQuantile(0.75))
+                .setField("p95", stats.getQuantile(0.95))
+                .setField("p98", stats.getQuantile(0.98))
+                .setField("p99", stats.getQuantile(0.99))
+                .setField("p999", stats.getQuantile(0.999));
+    }
+}

fluss-metrics/fluss-metrics-influxdb/src/main/java/org/apache/fluss/metrics/influxdb/InfluxdbReporter.java

@@ -0,0 +1,150 @@
+/*
+ * 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.influxdb;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metrics.Metric;
+import org.apache.fluss.metrics.groups.MetricGroup;
+import org.apache.fluss.metrics.reporter.ScheduledMetricReporter;
+import org.apache.fluss.utils.MapUtils;
+
+import com.influxdb.v3.client.InfluxDBClient;
+import com.influxdb.v3.client.Point;
+import com.influxdb.v3.client.config.ClientConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** {@link ScheduledMetricReporter} that exports {@link Metric Metrics} via InfluxDB. */
+public class InfluxdbReporter implements ScheduledMetricReporter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(InfluxdbReporter.class);
+
+    private static final char SCOPE_SEPARATOR = '_';
+    private static final String SCOPE_PREFIX = "fluss" + SCOPE_SEPARATOR;
+
+    private final Map<Metric, String> metricNames;
+    private final Map<Metric, List<Map.Entry<String, String>>> metricTags;
+    private final InfluxDBClient client;
+    private final Duration pushInterval;
+    private final InfluxdbPointProducer pointProducer;
+
+    public InfluxdbReporter(
+            String hostUrl, String org, String bucket, String token, Duration pushInterval) {
+        ClientConfig clientConfig =
+                new ClientConfig.Builder()
+                        .host(hostUrl)
+                        .token(token.toCharArray())
+                        .organization(org)
+                        .database(bucket)
+                        .build();
+
+        this.client = InfluxDBClient.getInstance(clientConfig);
+        this.pushInterval = pushInterval;
+        this.metricNames = MapUtils.newConcurrentHashMap();
+        this.metricTags = MapUtils.newConcurrentHashMap();
+        this.pointProducer = InfluxdbPointProducer.getInstance();
+
+        LOG.info("Started InfluxDB reporter connecting to {}", hostUrl);
+    }
+
+    @Override
+    public void open(Configuration config) {
+        // do nothing
+    }
+
+    @Override
+    public void close() {
+        if (client != null) {
+            try {
+                client.close();
+            } catch (Exception e) {
+                LOG.warn("Failed to close InfluxDB client", e);
+            }
+        }
+    }
+
+    @Override
+    public void report() {
+        List<Point> points = new ArrayList<>();
+        Instant now = Instant.now();
+
+        for (Map.Entry<Metric, String> entry : metricNames.entrySet()) {
+            Metric metric = entry.getKey();
+            String metricName = entry.getValue();
+            List<Map.Entry<String, String>> tags = metricTags.get(metric);
+
+            try {
+                Point point = pointProducer.createPoint(metric, metricName, tags, now);
+                points.add(point);
+            } catch (Exception e) {
+                LOG.warn("Failed to create point for metric {}", metricName, e);
+            }
+        }
+
+        client.writePoints(points);
+    }
+
+    @Override
+    public Duration scheduleInterval() {
+        return pushInterval;
+    }
+
+    @Override
+    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
+        String scopedMetricName = getScopedName(metricName, group);
+        List<Map.Entry<String, String>> tags = getTags(group);
+
+        metricNames.put(metric, scopedMetricName);
+        metricTags.put(metric, tags);
+    }
+
+    @Override
+    public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) {
+        metricNames.remove(metric);
+        metricTags.remove(metric);
+    }
+
+    private String getScopedName(String metricName, MetricGroup group) {
+        return SCOPE_PREFIX
+                + group.getLogicalScope(this::filterCharacters, SCOPE_SEPARATOR)
+                + SCOPE_SEPARATOR
+                + filterCharacters(metricName);
+    }
+
+    private List<Map.Entry<String, String>> getTags(MetricGroup group) {
+        List<Map.Entry<String, String>> tags = new ArrayList<>();
+        for (Map.Entry<String, String> entry : group.getAllVariables().entrySet()) {
+            tags.add(
+                    new HashMap.SimpleEntry<>(
+                            filterCharacters(entry.getKey()), filterCharacters(entry.getValue())));
+        }
+        return tags;
+    }
+
+    private String filterCharacters(String input) {
+        return input.replaceAll("[^a-zA-Z0-9_:]", String.valueOf(SCOPE_SEPARATOR));
+    }
+}