Merge pull request #183 from zhu-mingye/dev

获取Flink JobManager/TaskManagers相关信息

dlink-core/src/main/java/com/dlink/api/FlinkAPI.java

@@ -12,6 +12,7 @@ import com.dlink.gateway.result.SavePointResult;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -50,6 +51,16 @@ public class FlinkAPI {
         return parse(res);
     }
 
+    /**
+     *  get请求获取jobManger/TaskManager的日志 (结果为字符串并不是json格式)
+     * @param route
+     * @return
+     */
+    private String getResult(String route){
+        String res = HttpUtil.get(NetConstant.HTTP + address + NetConstant.SLASH + route, NetConstant.SERVER_TIME_OUT_ACTIVE);
+        return res;
+    }
+
     private JsonNode post(String route, String body) {
         String res = HttpUtil.post(NetConstant.HTTP + address + NetConstant.SLASH + route, body, NetConstant.SERVER_TIME_OUT_ACTIVE);
         return parse(res);
@@ -162,4 +173,51 @@ public class FlinkAPI {
     public JsonNode getJobsConfig(String jobId) {
         return get(FlinkRestAPIConstant.JOBS+jobId+FlinkRestAPIConstant.CONFIG);
     }
+
+    public JsonNode getJobManagerMetrics() {
+        return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.METRICS);
+    }
+
+    public JsonNode getJobManagerConfig() {
+        return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.CONFIG);
+    }
+
+    public String getJobManagerLog() {
+        return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOG);
+    }
+
+    public String getJobManagerStdOut() {
+        return getResult(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.STDOUT);
+    }
+
+    public JsonNode getJobManagerLogList() {
+        return get(FlinkRestAPIConstant.JOB_MANAGER + FlinkRestAPIConstant.LOGS);
+    }
+
+    public JsonNode getTaskManagers() {
+        return get(FlinkRestAPIConstant.TASK_MANAGER);
+    }
+
+    public JsonNode getTaskManagerMetrics(String containerId) {
+        return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.METRICS);
+    }
+
+
+    public String getTaskManagerLog(String containerId) {
+        return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.LOG);
+    }
+
+    public String getTaskManagerStdOut(String containerId) {
+        return getResult(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.STDOUT);
+    }
+
+    public JsonNode getTaskManagerLogList(String containerId) {
+        return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.LOGS);
+    }
+
+    public JsonNode getTaskManagerThreadDump(String containerId) {
+        return get(FlinkRestAPIConstant.TASK_MANAGER + containerId + FlinkRestAPIConstant.THREAD_DUMP);
+    }
+
+
 }

dlink-core/src/main/java/com/dlink/constant/FlinkRestAPIConstant.java

@@ -28,4 +28,18 @@ public final class FlinkRestAPIConstant {
 
     public static final String EXCEPTIONS = "/exceptions?maxExceptions=10";
 
+    public static final String JOB_MANAGER = "/jobmanager/";
+
+    public static final String TASK_MANAGER = "/taskmanagers/";
+
+    public static final String METRICS = "/metrics/";
+
+    public static final String LOG = "/log/";
+
+    public static final String LOGS = "/logs/";
+
+    public static final String STDOUT = "/stdout/";
+
+    public static final String THREAD_DUMP = "/thread-dump/";
+
 }

dlink-core/src/test/java/com/dlink/core/FlinkRestAPITest.java

@@ -16,7 +16,7 @@ import java.util.List;
 public class FlinkRestAPITest {
 
     //private String address = "192.168.123.157:8081";
-    private String address = "node02:45659";
+    private String address = "cdh5:8081";
 
     @Test
     public void savepointTest(){
@@ -35,4 +35,93 @@ public class FlinkRestAPITest {
     public void stopTest(){
         FlinkAPI.build(address).stop("0727f796fcf9e07d89e724f7e15598cf");
     }
+
+
+    @Test
+    public void getCheckPointsDetailInfoTest(){
+        JsonNode checkPointsDetailInfo = FlinkAPI.build(address).getCheckPointsConfig("178e954faaa4bf06cfbda971bb8b2957");
+        System.out.println(checkPointsDetailInfo.toString());
+    }
+
+    @Test
+    public void getConfigurationsDetailsInfoTest(){
+        JsonNode configurationsDetailsInfo = FlinkAPI.build(address).getJobsConfig("178e954faaa4bf06cfbda971bb8b2957");
+        System.out.println(configurationsDetailsInfo.toString());
+    }
+
+    @Test
+    public void getExectionsInfoTest(){
+        JsonNode exectionsDetailInfo = FlinkAPI.build(address).getException("178e954faaa4bf06cfbda971bb8b2957");
+        System.out.println(exectionsDetailInfo.toString());
+    }
+
+    @Test
+    public void getJobManagerMetricsTest(){
+        JsonNode jobManagerMetrics = FlinkAPI.build(address).getJobManagerMetrics();
+        System.out.println(jobManagerMetrics.toString());
+    }
+
+    @Test
+    public void getJobManagerConfigTest(){
+        JsonNode jobManagerConfig = FlinkAPI.build(address).getJobManagerConfig();
+        System.out.println(jobManagerConfig.toString());
+    }
+
+
+    @Test
+    public void getJobManagerLogTest(){
+        String jobManagerLog  = FlinkAPI.build(address).getJobManagerLog();
+        System.out.println(jobManagerLog);
+    }
+
+
+    @Test
+    public void getJobManagerStdOutTest(){
+        String jobManagerLogs = FlinkAPI.build(address).getJobManagerStdOut();
+        System.out.println(jobManagerLogs);
+    }
+
+    @Test
+    public void getJobManagerLogListTest(){
+        JsonNode jobManagerLogList = FlinkAPI.build(address).getJobManagerLogList();
+        System.out.println(jobManagerLogList.toString());
+    }
+
+    @Test
+    public void getTaskManagersTest(){
+        JsonNode taskManagers = FlinkAPI.build(address).getTaskManagers();
+        System.out.println(taskManagers.toString());
+    }
+
+    @Test
+    public void getTaskManagerMetricsTest(){
+        JsonNode taskManagerMetrics= FlinkAPI.build(address).getTaskManagerMetrics("container_e34_1646992539398_0004_01_000002");
+        System.out.println(taskManagerMetrics.toString());
+    }
+
+
+    @Test
+    public void getTaskManagerLogTest(){
+        String taskManagerLog= FlinkAPI.build(address).getTaskManagerLog("container_e34_1646992539398_0004_01_000002");
+        System.out.println(taskManagerLog);
+    }
+
+    @Test
+    public void getTaskManagerStdOutTest(){
+        String taskManagerStdOut = FlinkAPI.build(address).getTaskManagerStdOut("container_e34_1646992539398_0004_01_000002");
+        System.out.println(taskManagerStdOut);
+    }
+
+    @Test
+    public void getTaskManagerLogListTest(){
+        JsonNode taskManagerLogList= FlinkAPI.build(address).getTaskManagerLogList("container_e34_1646992539398_0004_01_000002");
+        System.out.println(taskManagerLogList.toString());
+    }
+
+
+    @Test
+    public void getTaskManagerThreadDumpTest(){
+        JsonNode taskManagerThreadDump= FlinkAPI.build(address).getTaskManagerThreadDump("container_e34_1646992539398_0004_01_000002");
+        System.out.println(taskManagerThreadDump.toString());
+    }
 }