feature(web): 增加 Overview 页面

方便总览全局情况，跨页面查看信息多有不便
2023-06-12 18:29:28 +08:00
parent d85c3a4864
commit 4e963fa537
8 changed files with 550 additions and 13 deletions
--- a/service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/OverviewController.java
+++ b/service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/OverviewController.java
@@ -0,0 +1,137 @@
+package com.lanyuanxiaoyao.service.web.controller;
+
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.util.StrUtil;
+import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
+import com.lanyuanxiaoyao.service.configuration.entity.AmisResponse;
+import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnApplication;
+import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnRootQueue;
+import com.lanyuanxiaoyao.service.forest.service.InfoService;
+import com.lanyuanxiaoyao.service.forest.service.QueueService;
+import com.lanyuanxiaoyao.service.forest.service.YarnService;
+import org.eclipse.collections.api.factory.Lists;
+import org.eclipse.collections.api.factory.Maps;
+import org.eclipse.collections.api.list.ImmutableList;
+import org.eclipse.collections.api.map.ImmutableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * 概览
+ *
+ * @author lanyuanxiaoyao
+ * @date 2023-06-12
+ */
+@RestController
+@RequestMapping("overview")
+public class OverviewController extends BaseController {
+    private static final Logger logger = LoggerFactory.getLogger(OverviewController.class);
+
+    private final InfoService infoService;
+    private final YarnService yarnService;
+    private final QueueService queueService;
+
+    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
+    public OverviewController(InfoService infoService, YarnService yarnService, QueueService queueService) {
+        this.infoService = infoService;
+        this.yarnService = yarnService;
+        this.queueService = queueService;
+    }
+
+    @GetMapping("")
+    public AmisResponse overview() throws ExecutionException, InterruptedException {
+        CompletableFuture<Long> tableTotalFuture = CompletableFuture.supplyAsync(infoService::tableTotal, ExecutorProvider.EXECUTORS);
+        CompletableFuture<Long> hudiTotalFuture = CompletableFuture.supplyAsync(infoService::hudiTotal, ExecutorProvider.EXECUTORS);
+        CompletableFuture<Long> focusCountFuture = CompletableFuture.supplyAsync(infoService::focusCount, ExecutorProvider.EXECUTORS);
+        CompletableFuture<Long> normalCountFuture = CompletableFuture.supplyAsync(infoService::normalCount, ExecutorProvider.EXECUTORS);
+        CompletableFuture.allOf(
+                tableTotalFuture,
+                hudiTotalFuture,
+                focusCountFuture,
+                normalCountFuture
+        );
+        return responseData()
+                .withData("table_total", tableTotalFuture.get())
+                .withData("hudi_total", hudiTotalFuture.get())
+                .withData("focus_count", focusCountFuture.get())
+                .withData("normal_count", normalCountFuture.get());
+    }
+
+    @GetMapping("yarn-job")
+    private ImmutableMap<String, Object> yarnOverview(@RequestParam("cluster") String cluster, @RequestParam("search") String text) {
+        boolean isSearch = StrUtil.isNotBlank(text);
+        ImmutableList<YarnApplication> applications = yarnService.jobList(cluster).select(app -> !isSearch || StrUtil.contains(app.getName(), text));
+        return Maps.immutable.ofAll(MapUtil.<String, Object>builder()
+                .put("name", cluster)
+                .put("total", applications.size())
+                .put("running", applications.count(app -> StrUtil.equals(app.getState(), "RUNNING")))
+                .put("scheduling", applications.count(app -> StrUtil.equals(app.getState(), "ACCEPTED")))
+                .put("failure", applications.count(app -> StrUtil.equals(app.getState(), "FAILED")))
+                .build());
+    }
+
+    @GetMapping("yarn")
+    public AmisResponse yarnOverview(@RequestParam("clusters") List<String> clusters, @RequestParam("search") String text) {
+        ImmutableList<ImmutableMap<String, Object>> maps = Lists.immutable.ofAll(clusters)
+                .asParallel(ExecutorProvider.EXECUTORS, 1)
+                .collect(cluster -> yarnOverview(cluster, text))
+                .toList()
+                .toImmutable();
+        long total = maps.sumOfInt(m -> (int) m.get("total"));
+        long running = maps.sumOfInt(m -> (int) m.get("running"));
+        long scheduling = maps.sumOfInt(m -> (int) m.get("scheduling"));
+        long failure = maps.sumOfInt(m -> (int) m.get("failure"));
+        return responseData()
+                .withData("total", total)
+                .withData("running", running)
+                .withData("scheduling", scheduling)
+                .withData("failure", failure)
+                .withData("items", maps);
+    }
+
+    @GetMapping("yarn-cluster")
+    public AmisResponse yarnClusterOverview(@RequestParam("cluster") String cluster, @RequestParam("queue") String queue) {
+        AmisResponse response = responseData();
+        YarnRootQueue root = yarnService.cluster(cluster);
+        response.withData("root", root);
+        if (StrUtil.isNotBlank(queue)) {
+            yarnService.queueList(cluster)
+                    .select(q -> StrUtil.equals(q.getQueueName(), queue))
+                    .getFirstOptional()
+                    .ifPresent(yarnQueue -> response.withData("target", yarnQueue));
+        }
+        return response;
+    }
+
+    @GetMapping("queue")
+    public AmisResponse queueOverview(@RequestParam("queue") String queue) {
+        return responseData()
+                .withData("size", queueService.size(queue));
+    }
+
+    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
+
+    @GetMapping("version")
+    public AmisResponse versionOverview() throws ExecutionException, InterruptedException {
+        String version = LocalDateTime.now().minusDays(1).format(FORMATTER);
+        CompletableFuture<Long> unReceiveNormalTableCount = CompletableFuture.supplyAsync(() -> infoService.unReceiveVersionNormalTableCount(version), ExecutorProvider.EXECUTORS);
+        CompletableFuture<Long> unReceiveFocusCount = CompletableFuture.supplyAsync(() -> infoService.unReceiveVersionFocusTableCount(version), ExecutorProvider.EXECUTORS);
+        CompletableFuture<Long> unScheduledNormalTableCount = CompletableFuture.supplyAsync(() -> infoService.unScheduledNormalTableCount(version), ExecutorProvider.EXECUTORS);
+        CompletableFuture<Long> unScheduledFocusTableCount = CompletableFuture.supplyAsync(() -> infoService.unScheduledFocusTableCount(version), ExecutorProvider.EXECUTORS);
+        CompletableFuture.allOf(unReceiveNormalTableCount, unReceiveFocusCount, unScheduledNormalTableCount, unScheduledFocusTableCount);
+        return responseData()
+                .withData("version", version)
+                .withData("unReceive", Maps.immutable.of("normal", unReceiveNormalTableCount.get(), "focus", unReceiveFocusCount.get()))
+                .withData("unSchedule", Maps.immutable.of("normal", unScheduledNormalTableCount.get(), "focus", unScheduledFocusTableCount.get()));
+    }
+}