[metric] Support influxdb reporter

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

@@ -2027,6 +2027,50 @@ 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_VERSION =
+            key("metrics.reporter.influxdb.version")
+                    .stringType()
+                    .defaultValue("v3")
+                    .withDescription(
+                            "The InfluxDB version to connect to. Supported values are 'v2' and 'v3'. "
+                                    + "For InfluxDB v2, the 'org' configuration is required; for InfluxDB v3, it is not needed.");
+
+    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. Required for InfluxDB v2, not needed for InfluxDB v3.");
+
+    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-dist/pom.xml

@@ -76,6 +76,13 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-metrics-influxdb</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.fluss</groupId>
             <artifactId>fluss-lake-paimon</artifactId>

fluss-dist/src/main/assemblies/plugins.xml

@@ -80,6 +80,13 @@
             <fileMode>0644</fileMode>
         </file>
 
+        <file>
+            <source>../fluss-metrics/fluss-metrics-influxdb/target/fluss-metrics-influxdb-${project.version}.jar</source>
+            <outputDirectory>plugins/influxdb/</outputDirectory>
+            <destName>fluss-metrics-influxdb-${project.version}.jar</destName>
+            <fileMode>0644</fileMode>
+        </file>
+
         <!-- lake formats -->
         <file>
             <source>../fluss-lake/fluss-lake-paimon/target/fluss-lake-paimon-${project.version}.jar</source>

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

@@ -0,0 +1,154 @@
+<?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>1.0-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>
+        </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>
+
+        <dependency>
+            <groupId>com.github.tomakehurst</groupId>
+            <artifactId>wiremock-jre8</artifactId>
+            <version>2.32.0</version>
+            <scope>test</scope>
+        </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>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>shade-fluss</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <artifactSet>
+                                <includes>
+                                    <include>com.influxdb:*</include>
+                                    <include>org.apache.arrow:*</include>
+                                    <include>io.grpc:*</include>
+                                    <include>io.netty:*</include>
+                                    <include>com.google.protobuf:*</include>
+                                    <include>com.google.guava:*</include>
+                                    <include>com.fasterxml.jackson.core:*</include>
+                                    <include>com.google.flatbuffers:*</include>
+                                    <include>io.perfmark:*</include>
+                                    <include>com.google.code.findbugs:*</include>
+                                    <include>org.codehaus.mojo:*</include>
+                                </includes>
+                            </artifactSet>
+                            <relocations>
+                                <relocation>
+                                    <pattern>com.influxdb</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.com.influxdb</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>org.apache.arrow</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.org.apache.arrow</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>io.grpc</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.io.grpc</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>io.netty</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.io.netty</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.google.protobuf</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.com.google.protobuf</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.google.common</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.com.google.common</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.fasterxml.jackson</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.com.fasterxml.jackson</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.google.flatbuffers</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.com.google.flatbuffers</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>org.codehaus.mojo.annotations</pattern>
+                                    <shadedPattern>org.apache.fluss.metrics.influxdb.shaded.org.codehaus.mojo.annotations</shadedPattern>
+                                </relocation>
+                            </relocations>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

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

@@ -0,0 +1,106 @@
+/*
+ * 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);
+
+        if (tags != null) {
+            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).booleanValue());
+        } 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));
+    }
+}