[Metrics] Add OpenTelemetry metrics exporter

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

@@ -1653,6 +1653,42 @@ public class ConfigOptions {
                                     + "the CoordinatorServer) it is advisable to use a port range "
                                     + "like 9990-9999.");
 
+    // ------------------------------------------------------------------------
+    //  ConfigOptions for OpenTelemetry reporter
+    // ------------------------------------------------------------------------
+    public static final ConfigOption<String> METRICS_REPORTER_OPENTELEMETRY_ENDPOINT =
+            key("metrics.reporter.opentelemetry.endpoint")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Target to which the OpenTelemetry metric reporter is going to send metrics to.");
+
+    public static final ConfigOption<Duration> METRICS_REPORTER_OPENTELEMETRY_EXPORT_INTERVAL =
+            key("metrics.reporter.opentelemetry.export-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(10))
+                    .withDescription(
+                            "Frequency of metric export by the OpenTelemetry metric reporter to the endpoint.");
+
+    public static final ConfigOption<Duration> METRICS_REPORTER_OPENTELEMETRY_EXPORT_TIMEOUT =
+            key("metrics.reporter.opentelemetry.export-timeout")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(10))
+                    .withDescription(
+                            "Maximum time the OpenTelemetry metric reporter will wait for each metric export.");
+
+    public static final ConfigOption<String> METRICS_REPORTER_OPENTELEMETRY_SERVICE_NAME =
+            key("metrics.reporter.opentelemetry.service.name")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Service name that is set in the OpenTelemetry Resource.");
+
+    public static final ConfigOption<String> METRICS_REPORTER_OPENTELEMETRY_SERVICE_VERSION =
+            key("metrics.reporter.opentelemetry.service.version")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Service version that is set in the OpenTelemetry Resource.");
+
     // ------------------------------------------------------------------------
     //  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-opentelemetry</artifactId>
+            <version>0.8-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.fluss</groupId>
             <artifactId>fluss-lake-paimon</artifactId>
@@ -96,7 +103,7 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
-      
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-shaded-hadoop-2-uber</artifactId>

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

@@ -81,6 +81,15 @@
             <fileMode>0644</fileMode>
         </file>
 
+        <file>
+            <source>
+                ../fluss-metrics/fluss-metrics-opentelemetry/target/fluss-metrics-opentelemetry-${project.version}.jar
+            </source>
+            <outputDirectory>plugins/opentelemetry/</outputDirectory>
+            <destName>fluss-metrics-opentelemetry-${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-opentelemetry/pom.xml

@@ -0,0 +1,117 @@
+<?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.8-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>fluss-metrics-opentelemetry</artifactId>
+    <name>Fluss : Metrics : OpenTelemetry</name>
+
+    <properties>
+        <opentelemetry.version>1.51.0</opentelemetry.version>
+        <opentelemetry-semconv.version>1.34.0</opentelemetry-semconv.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-common</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk-metrics</artifactId>
+            <version>${opentelemetry.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-otlp</artifactId>
+            <version>${opentelemetry.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry.semconv</groupId>
+            <artifactId>opentelemetry-semconv</artifactId>
+            <version>${opentelemetry-semconv.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>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <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>*:*</include>
+                                </includes>
+                            </artifactSet>
+                            <filters>
+                                <filter>
+                                    <artifact>*</artifact>
+                                    <excludes>
+                                        <exclude>okhttp3/internal/publicsuffix/NOTICE</exclude>
+                                    </excludes>
+                                </filter>
+                            </filters>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

fluss-metrics/fluss-metrics-opentelemetry/src/main/java/org/apache/fluss/metrics/opentelemetry/MetricMetadata.java

@@ -0,0 +1,55 @@
+/*
+ * 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.opentelemetry;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+
+import java.util.Map;
+
+/* This file is based on source code of Apache Flink Project (https://flink.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Metadata associated with a given metric, stored alongside it so that it can be provided during
+ * reporting time.
+ */
+class MetricMetadata {
+
+    @VisibleForTesting static final String NAME_SEPARATOR = ".";
+
+    private final String name;
+    private final Map<String, String> variables;
+
+    public MetricMetadata(String name, Map<String, String> variables) {
+        this.name = name;
+        this.variables = variables;
+    }
+
+    public MetricMetadata subMetric(String suffix) {
+        return new MetricMetadata(name + NAME_SEPARATOR + suffix, variables);
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public Map<String, String> getVariables() {
+        return variables;
+    }
+}