[improve][API] support metrics interface in rest v2 (#8312)

zilliztech · Dec 25, 2024 · bd89d84 · bd89d84
1 parent eae32bf
commit bd89d84
Show file tree

Hide file tree

Showing 17 changed files with 382 additions and 183 deletions.
diff --git a/docs/en/seatunnel-engine/rest-api-v2.md b/docs/en/seatunnel-engine/rest-api-v2.md
@@ -845,3 +845,18 @@ To get a list of logs from the current node: `http://localhost:5801/log`
 To get the content of a log file: `http://localhost:5801/log/job-898380162133917698.log`
 
 </details>
+
+
+### Get Node Metrics
+
+<details>
+ <summary>
+    <code>GET</code> <code><b>/metrics</b></code>  
+    <code>GET</code> <code><b>/openmetrics</b></code>
+</summary>
+
+To get the metrics, you need to open `Telemetry` first, or you will get an empty response.  
+
+More information about `Telemetry` can be found in the [Telemetry](telemetry.md) documentation.
+
+</details>
diff --git a/docs/zh/seatunnel-engine/rest-api-v2.md b/docs/zh/seatunnel-engine/rest-api-v2.md
@@ -850,3 +850,16 @@ sink {
 获取日志文件内容：`http://localhost:5801/log/job-898380162133917698.log``
 
 </details>
+
+### 获取节点指标信息
+
+<details>
+ <summary>
+    <code>GET</code> <code><b>/metrics</b></code>  
+    <code>GET</code> <code><b>/openmetrics</b></code>
+</summary>
+你需要先打开`Telemetry`才能获取集群指标信息。否则将返回空信息。
+
+更多关于`Telemetry`的信息可以在[Telemetry](telemetry.md)文档中找到。
+
+</details>
diff --git a/pom.xml b/pom.xml
@@ -122,6 +122,7 @@
         <jcommander.version>1.81</jcommander.version>
         <junit4.version>4.13.2</junit4.version>
         <junit5.version>5.9.0</junit5.version>
+        <rest-assured.version>5.4.0</rest-assured.version>
         <mockito.version>4.11.0</mockito.version>
         <config.version>1.3.3</config.version>
         <maven-shade-plugin.version>3.4.1</maven-shade-plugin.version>

diff --git a/...e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterSeaTunnelEngineContainer.java b/...e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterSeaTunnelEngineContainer.java
diff --git a/...seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/MultiTableMetricsIT.java b/...seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/MultiTableMetricsIT.java
@@ -91,7 +91,7 @@ public void multiTableMetrics() {
                                                                     .getAddress()
                                                                     .getPort()
                                                             + RestConstant.CONTEXT_PATH
-                                                            + RestConstant.JOB_INFO_URL
+                                                            + RestConstant.REST_URL_JOB_INFO
                                                             + "/"
                                                             + batchJobProxy.getJobId());
                             // In the test example, the data size of a single [3, "C", 100] is 13

diff --git a/...connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java b/...connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/RestApiIT.java
diff --git a/...e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java b/...e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java
@@ -136,7 +136,7 @@ public void testFakeSourceToInMemorySinkForRestApi() throws IOException, Interru
                                     + colon
                                     + server.getFirstMappedPort()
                                     + RestConstant.CONTEXT_PATH
-                                    + RestConstant.SUBMIT_JOB_URL)
+                                    + RestConstant.REST_URL_SUBMIT_JOB)
                     .then()
                     .statusCode(200);
 
@@ -150,7 +150,7 @@ public void testFakeSourceToInMemorySinkForRestApi() throws IOException, Interru
                                                             + colon
                                                             + server.getFirstMappedPort()
                                                             + RestConstant.CONTEXT_PATH
-                                                            + RestConstant.FINISHED_JOBS_INFO
+                                                            + RestConstant.REST_URL_FINISHED_JOBS
                                                             + "/FINISHED")
                                             .then()
                                             .statusCode(200)

diff --git a/...tor-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/joblog/JobLogIT.java b/...tor-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/joblog/JobLogIT.java
@@ -269,13 +269,13 @@ private Response submitJob(
                                                 + ":"
                                                 + container.getFirstMappedPort()
                                                 + RestConstant.CONTEXT_PATH
-                                                + RestConstant.SUBMIT_JOB_URL
+                                                + RestConstant.REST_URL_SUBMIT_JOB
                                         : "http://"
                                                 + container.getHost()
                                                 + ":"
                                                 + container.getFirstMappedPort()
                                                 + RestConstant.CONTEXT_PATH
-                                                + RestConstant.SUBMIT_JOB_URL
+                                                + RestConstant.REST_URL_SUBMIT_JOB
                                                 + "?"
                                                 + parameters);
         return response;

diff --git a/...rg/apache/seatunnel/engine/e2e/telemetry/MasterWorkerClusterSeaTunnelWithTelemetryIT.java b/...rg/apache/seatunnel/engine/e2e/telemetry/MasterWorkerClusterSeaTunnelWithTelemetryIT.java
@@ -121,7 +121,7 @@ public void startUp() throws Exception {
                                                     + colon
                                                     + server.getFirstMappedPort()
                                                     + CONTEXT_PATH
-                                                    + RestConstant.JOB_INFO_URL
+                                                    + RestConstant.REST_URL_JOB_INFO
                                                     + "/"
                                                     + JobId)
                                     .then()
@@ -652,13 +652,13 @@ private Response submitJob(
                                                 + colon
                                                 + port
                                                 + contextPath
-                                                + RestConstant.SUBMIT_JOB_URL
+                                                + RestConstant.REST_URL_SUBMIT_JOB
                                         : http
                                                 + container.getHost()
                                                 + colon
                                                 + port
                                                 + contextPath
-                                                + RestConstant.SUBMIT_JOB_URL
+                                                + RestConstant.REST_URL_SUBMIT_JOB
                                                 + "?"
                                                 + parameters);
         return response;

diff --git a/seatunnel-engine/seatunnel-engine-server/pom.xml b/seatunnel-engine/seatunnel-engine-server/pom.xml
@@ -108,6 +108,19 @@
             <classifier>optional</classifier>
             <scope>provided</scope>
         </dependency>
+
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <version>${rest-assured.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.codehaus.groovy</groupId>
+                    <artifactId>groovy</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
             <artifactId>seatunnel-e2e-common</artifactId>

diff --git a/...eatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java b/...eatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java
@@ -31,6 +31,7 @@
 import org.apache.seatunnel.engine.server.rest.servlet.EncryptConfigServlet;
 import org.apache.seatunnel.engine.server.rest.servlet.FinishedJobsServlet;
 import org.apache.seatunnel.engine.server.rest.servlet.JobInfoServlet;
+import org.apache.seatunnel.engine.server.rest.servlet.MetricsServlet;
 import org.apache.seatunnel.engine.server.rest.servlet.OverviewServlet;
 import org.apache.seatunnel.engine.server.rest.servlet.RunningJobsServlet;
 import org.apache.seatunnel.engine.server.rest.servlet.RunningThreadsServlet;
@@ -53,23 +54,25 @@
 import java.net.URL;
 import java.util.EnumSet;
 
-import static org.apache.seatunnel.engine.server.rest.RestConstant.ENCRYPT_CONFIG;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.FINISHED_JOBS_INFO;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.GET_ALL_LOG_NAME;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.GET_LOG;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.GET_LOGS;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.JOB_INFO_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.OVERVIEW;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.RUNNING_JOBS_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.RUNNING_JOB_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.RUNNING_THREADS;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.STOP_JOBS_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.STOP_JOB_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.SUBMIT_JOBS_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.SUBMIT_JOB_URL;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.SYSTEM_MONITORING_INFORMATION;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.THREAD_DUMP;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.UPDATE_TAGS_URL;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_ENCRYPT_CONFIG;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_FINISHED_JOBS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_GET_ALL_LOG_NAME;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_JOB_INFO;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_LOG;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_LOGS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_METRICS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_OPEN_METRICS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_OVERVIEW;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_RUNNING_JOB;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_RUNNING_JOBS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_RUNNING_THREADS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_STOP_JOB;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_STOP_JOBS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SUBMIT_JOB;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SUBMIT_JOBS;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SYSTEM_MONITORING_INFORMATION;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_THREAD_DUMP;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_UPDATE_TAGS;
 
 /** The Jetty service for SeaTunnel engine server. */
 @Slf4j
@@ -134,26 +137,31 @@ public void createJettyServer() {
                 new ServletHolder(new CurrentNodeLogServlet(nodeEngine));
         ServletHolder allLogNameServlet = new ServletHolder(new AllLogNameServlet(nodeEngine));
 
-        context.addServlet(overviewHolder, convertUrlToPath(OVERVIEW));
-        context.addServlet(runningJobsHolder, convertUrlToPath(RUNNING_JOBS_URL));
-        context.addServlet(finishedJobsHolder, convertUrlToPath(FINISHED_JOBS_INFO));
-        context.addServlet(systemMonitoringHolder, convertUrlToPath(SYSTEM_MONITORING_INFORMATION));
-        context.addServlet(jobInfoHolder, convertUrlToPath(JOB_INFO_URL));
-        context.addServlet(jobInfoHolder, convertUrlToPath(RUNNING_JOB_URL));
-        context.addServlet(threadDumpHolder, convertUrlToPath(THREAD_DUMP));
-
-        context.addServlet(submitJobHolder, convertUrlToPath(SUBMIT_JOB_URL));
-        context.addServlet(submitJobsHolder, convertUrlToPath(SUBMIT_JOBS_URL));
-        context.addServlet(stopJobHolder, convertUrlToPath(STOP_JOB_URL));
-        context.addServlet(stopJobsHolder, convertUrlToPath(STOP_JOBS_URL));
-        context.addServlet(encryptConfigHolder, convertUrlToPath(ENCRYPT_CONFIG));
-        context.addServlet(updateTagsHandler, convertUrlToPath(UPDATE_TAGS_URL));
-
-        context.addServlet(runningThreadsHolder, convertUrlToPath(RUNNING_THREADS));
-
-        context.addServlet(allNodeLogServletHolder, convertUrlToPath(GET_LOGS));
-        context.addServlet(currentNodeLogServlet, convertUrlToPath(GET_LOG));
-        context.addServlet(allLogNameServlet, convertUrlToPath(GET_ALL_LOG_NAME));
+        ServletHolder metricsServlet = new ServletHolder(new MetricsServlet(nodeEngine));
+
+        context.addServlet(overviewHolder, convertUrlToPath(REST_URL_OVERVIEW));
+        context.addServlet(runningJobsHolder, convertUrlToPath(REST_URL_RUNNING_JOBS));
+        context.addServlet(finishedJobsHolder, convertUrlToPath(REST_URL_FINISHED_JOBS));
+        context.addServlet(
+                systemMonitoringHolder, convertUrlToPath(REST_URL_SYSTEM_MONITORING_INFORMATION));
+        context.addServlet(jobInfoHolder, convertUrlToPath(REST_URL_JOB_INFO));
+        context.addServlet(jobInfoHolder, convertUrlToPath(REST_URL_RUNNING_JOB));
+        context.addServlet(threadDumpHolder, convertUrlToPath(REST_URL_THREAD_DUMP));
+
+        context.addServlet(submitJobHolder, convertUrlToPath(REST_URL_SUBMIT_JOB));
+        context.addServlet(submitJobsHolder, convertUrlToPath(REST_URL_SUBMIT_JOBS));
+        context.addServlet(stopJobHolder, convertUrlToPath(REST_URL_STOP_JOB));
+        context.addServlet(stopJobsHolder, convertUrlToPath(REST_URL_STOP_JOBS));
+        context.addServlet(encryptConfigHolder, convertUrlToPath(REST_URL_ENCRYPT_CONFIG));
+        context.addServlet(updateTagsHandler, convertUrlToPath(REST_URL_UPDATE_TAGS));
+
+        context.addServlet(runningThreadsHolder, convertUrlToPath(REST_URL_RUNNING_THREADS));
+
+        context.addServlet(allNodeLogServletHolder, convertUrlToPath(REST_URL_LOGS));
+        context.addServlet(currentNodeLogServlet, convertUrlToPath(REST_URL_LOG));
+        context.addServlet(allLogNameServlet, convertUrlToPath(REST_URL_GET_ALL_LOG_NAME));
+        context.addServlet(metricsServlet, convertUrlToPath(REST_URL_METRICS));
+        context.addServlet(metricsServlet, convertUrlToPath(REST_URL_OPEN_METRICS));
 
         server.setHandler(context);
 

diff --git a/...nel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java b/...nel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java
@@ -60,34 +60,33 @@ public class RestConstant {
     public static final String TABLE_SINK_WRITE_BYTES_PER_SECONDS = "TableSinkWriteBytesPerSeconds";
 
     public static final String CONTEXT_PATH = "/hazelcast/rest/maps";
-
-    public static final String OVERVIEW = "/overview";
-
-    public static final String RUNNING_JOBS_URL = "/running-jobs";
-    @Deprecated public static final String RUNNING_JOB_URL = "/running-job";
-    public static final String JOB_INFO_URL = "/job-info";
-    public static final String FINISHED_JOBS_INFO = "/finished-jobs";
-    public static final String ENCRYPT_CONFIG = "/encrypt-config";
-    public static final String THREAD_DUMP = "/thread-dump";
-
+    public static final String INSTANCE_CONTEXT_PATH = "/hazelcast/rest/instance";
+
+    // api path start
+    public static final String REST_URL_OVERVIEW = "/overview";
+    public static final String REST_URL_RUNNING_JOBS = "/running-jobs";
+    @Deprecated public static final String REST_URL_RUNNING_JOB = "/running-job";
+    public static final String REST_URL_JOB_INFO = "/job-info";
+    public static final String REST_URL_FINISHED_JOBS = "/finished-jobs";
+    public static final String REST_URL_ENCRYPT_CONFIG = "/encrypt-config";
+    public static final String REST_URL_THREAD_DUMP = "/thread-dump";
     // only for test use
-    public static final String RUNNING_THREADS = "/running-threads";
-
-    public static final String SYSTEM_MONITORING_INFORMATION = "/system-monitoring-information";
-
-    public static final String SUBMIT_JOB_URL = "/submit-job";
-    public static final String SUBMIT_JOBS_URL = "/submit-jobs";
-    public static final String STOP_JOB_URL = "/stop-job";
-    public static final String STOP_JOBS_URL = "/stop-jobs";
-    public static final String UPDATE_TAGS_URL = "/update-tags";
+    public static final String REST_URL_RUNNING_THREADS = "/running-threads";
+    public static final String REST_URL_SYSTEM_MONITORING_INFORMATION =
+            "/system-monitoring-information";
+    public static final String REST_URL_SUBMIT_JOB = "/submit-job";
+    public static final String REST_URL_SUBMIT_JOBS = "/submit-jobs";
+    public static final String REST_URL_STOP_JOB = "/stop-job";
+    public static final String REST_URL_STOP_JOBS = "/stop-jobs";
+    public static final String REST_URL_UPDATE_TAGS = "/update-tags";
     // Get All Nodes Log
-    public static final String GET_LOGS = "/logs";
+    public static final String REST_URL_LOGS = "/logs";
     // Get Current Node Log
-    public static final String GET_LOG = "/log";
+    public static final String REST_URL_LOG = "/log";
     // Code internal Use , Get Node Log Name
-    public static final String GET_ALL_LOG_NAME = "/get-all-log-name";
+    public static final String REST_URL_GET_ALL_LOG_NAME = "/get-all-log-name";
+    public static final String REST_URL_METRICS = "/metrics";
+    public static final String REST_URL_OPEN_METRICS = "/openmetrics";
+    // api path end
 
-    // metrics
-    public static final String TELEMETRY_METRICS_URL = "/hazelcast/rest/instance/metrics";
-    public static final String TELEMETRY_OPEN_METRICS_URL = "/hazelcast/rest/instance/openmetrics";
 }