Add Flink version to Garmadon logs (#236)

Co-authored-by: Jean-Baptiste Catté <jb.catte@criteo.com>
criteo · Feb 28, 2025 · 4db8bf5 · 4db8bf5
1 parent 885b878
commit 4db8bf5
Show file tree

Hide file tree

Showing 5 changed files with 47 additions and 1 deletion.
diff --git a/agent/src/main/java/com/criteo/hadoop/garmadon/agent/headers/ContainerHeader.java b/agent/src/main/java/com/criteo/hadoop/garmadon/agent/headers/ContainerHeader.java
@@ -1,5 +1,6 @@
 package com.criteo.hadoop.garmadon.agent.headers;
 
+import com.criteo.hadoop.garmadon.jvm.utils.FlinkRuntime;
 import com.criteo.hadoop.garmadon.jvm.utils.JavaRuntime;
 import com.criteo.hadoop.garmadon.jvm.utils.SparkRuntime;
 import com.criteo.hadoop.garmadon.schema.enums.Component;
@@ -80,11 +81,13 @@ private void setFrameworkComponent() {
             case "org.apache.flink.yarn.entrypoint.YarnJobClusterEntrypoint":
             case "org.apache.flink.yarn.entrypoint.YarnSessionClusterEntrypoint":
                 framework = Framework.FLINK;
+                frameworkVersion = FlinkRuntime.getVersion();
                 component = Component.APP_MASTER;
                 break;
             case "org.apache.flink.yarn.YarnTaskManager":
             case "org.apache.flink.yarn.YarnTaskExecutorRunner":
                 framework = Framework.FLINK;
+                frameworkVersion = FlinkRuntime.getVersion();
                 component = Component.TASK_MANAGER;
                 break;
             // YARN

diff --git a/jvm-statistics/pom.xml b/jvm-statistics/pom.xml
@@ -63,5 +63,11 @@
             <version>3.2.1</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime_2.12</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/jvm-statistics/src/main/java/com/criteo/hadoop/garmadon/jvm/utils/FlinkRuntime.java b/jvm-statistics/src/main/java/com/criteo/hadoop/garmadon/jvm/utils/FlinkRuntime.java
@@ -0,0 +1,24 @@
+package com.criteo.hadoop.garmadon.jvm.utils;
+
+import java.lang.reflect.Method;
+
+public final class FlinkRuntime {
+    private static final String VERSION = computeVersion();
+
+    private FlinkRuntime() {
+    }
+
+    public static String getVersion() {
+        return VERSION;
+    }
+
+    static String computeVersion() {
+        try {
+            Class<?> clazz = Class.forName("org.apache.flink.runtime.util.EnvironmentInformation");
+            Method versionMethod = clazz.getDeclaredMethod("getVersion");
+            return (String) versionMethod.invoke(null);
+        } catch (Throwable e) {
+            return "unknown";
+        }
+    }
+}
diff --git a/jvm-statistics/src/test/java/com/criteo/hadoop/garmadon/jvm/utils/FlinkRuntimeTest.java b/jvm-statistics/src/test/java/com/criteo/hadoop/garmadon/jvm/utils/FlinkRuntimeTest.java
@@ -0,0 +1,13 @@
+package com.criteo.hadoop.garmadon.jvm.utils;
+
+import junit.framework.TestCase;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class FlinkRuntimeTest extends TestCase {
+
+    public void test_get_version() {
+        assertThat(FlinkRuntime.getVersion()).isEqualTo("1.9.3");
+    }
+
+}
diff --git a/pom.xml b/pom.xml
@@ -106,7 +106,7 @@
         <protobuf-dynamic.version>1.0.0</protobuf-dynamic.version>
         <jackson.version>2.9.9</jackson.version>
         <commons-io.version>2.6</commons-io.version>
-        <flink.version>1.6.4</flink.version>
+        <flink.version>1.9.3</flink.version>
         <prestodb.version>0.217</prestodb.version>
         <prestosql.version>315</prestosql.version>
         <assertj.version>3.11.1</assertj.version>