lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(web): 优化统一的返回值封装和适配

Browse Source
This commit is contained in:
兰缘小妖
2023-07-06 14:31:24 +08:00
parent 47bd555fbf
commit 894faae2ed
18 changed files with 288 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/CloudController.java
View File

@@ -3,6 +3,8 @@ package com.lanyuanxiaoyao.service.web.controller;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisMapResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.CloudServiceVO;
@@ -81,19 +83,19 @@ public class CloudController extends BaseController {
}
@GetMapping("list")
public AmisResponse list() {
public AmisCrudResponse list() {
ImmutableList<CloudServiceVO> serviceVOS = serviceVOS(CloudServiceVO.Service::getServiceId);
return AmisResponse.responseCrudData(serviceVOS, serviceVOS.size());
}
@GetMapping("list_ip")
public AmisResponse listIp() {
public AmisCrudResponse listIp() {
ImmutableList<CloudServiceVO> serviceVOS = serviceVOS(CloudServiceVO.Service::getHost);
return AmisResponse.responseCrudData(serviceVOS, serviceVOS.size());
}
@GetMapping("deploy_plan")
public AmisResponse deployPlan() {
public AmisMapResponse deployPlan() {
ImmutableList<CloudServiceVO.Service> services = services();
ImmutableList<String> deployExists = services.collect(service -> StrUtil.format("{}-{}", service.getServiceId(), service.getHost()));
ImmutableList<String> serviceIds = services.collect(CloudServiceVO.Service::getServiceId).distinct().toSortedList().toImmutable();
@@ -122,7 +124,7 @@ public class CloudController extends BaseController {
.put("tpl", "${" + key + "}")
.build();
}));
return AmisResponse.responseData(
return AmisResponse.responseMapData(
MapUtil.<String, Object>builder()
.put("type", "service")
.put("data", MapUtil.builder()
@@ -140,7 +142,7 @@ public class CloudController extends BaseController {
}
@GetMapping("heart")
public AmisResponse heart() {
return AmisResponse.responseData().withData("status", true);
public AmisMapResponse heart() {
return AmisResponse.responseMapData("status", true);
}
}

10
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/FlinkController.java
View File

@@ -2,9 +2,11 @@ package com.lanyuanxiaoyao.service.web.controller;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.flink.FlinkCheckpoint;
import com.lanyuanxiaoyao.service.configuration.entity.flink.FlinkOverview;
import com.lanyuanxiaoyao.service.forest.service.FlinkService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.FlinkVertexVO;
import java.util.concurrent.ExecutionException;
@@ -35,12 +37,12 @@ public class FlinkController extends BaseController {
}
@GetMapping("overview")
public AmisResponse overview(@RequestParam("url") String url) throws ExecutionException, InterruptedException {
return AmisResponse.responseDetail(flinkService.overview(url));
public AmisResponse<FlinkOverview> overview(@RequestParam("url") String url) throws ExecutionException, InterruptedException {
return AmisResponse.responseSuccess(flinkService.overview(url));
}
@GetMapping("jobs")
public AmisResponse jobs(
public AmisCrudResponse jobs(
@RequestParam("url") String url,
@RequestParam(value = "mode", required = false) String mode,
@RequestParam(value = "schema", required = false) String schema,

24
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/HudiController.java
View File

@@ -2,10 +2,12 @@ package com.lanyuanxiaoyao.service.web.controller;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.PageResponse;
import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiCompactionPlan;
import com.lanyuanxiaoyao.service.configuration.entity.hudi.HudiInstant;
import com.lanyuanxiaoyao.service.forest.service.HudiService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import java.util.List;
import java.util.function.Function;
@@ -29,7 +31,10 @@ import org.springframework.web.bind.annotation.RestController;
@RequestMapping("hudi")
public class HudiController extends BaseController {
private static final Logger logger = LoggerFactory.getLogger(HudiController.class);
private static final ImmutableMap<String, Function<HudiInstant, String>> TIMELINE_SORT_MAP = Maps.immutable.of(
"timestamp",
HudiInstant::getTimestamp
);
private final HudiService hudiService;
@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@@ -37,13 +42,8 @@ public class HudiController extends BaseController {
this.hudiService = hudiService;
}
private static final ImmutableMap<String, Function<HudiInstant, String>> TIMELINE_SORT_MAP = Maps.immutable.of(
"timestamp",
HudiInstant::getTimestamp
);
@GetMapping("/timeline/list")
public AmisResponse timeline(
public AmisCrudResponse timeline(
@RequestParam(value = "page", defaultValue = "1") Integer page,
@RequestParam(value = "count", defaultValue = "10") Integer count,
@RequestParam(value = "order", defaultValue = "timestamp") String order,
@@ -69,7 +69,7 @@ public class HudiController extends BaseController {
}
@GetMapping("/timeline/list_hdfs")
public AmisResponse timelineHdfs(
public AmisCrudResponse timelineHdfs(
@RequestParam(value = "page", defaultValue = "1") Integer page,
@RequestParam(value = "count", defaultValue = "10") Integer count,
@RequestParam(value = "order", defaultValue = "timestamp") String order,
@@ -103,16 +103,16 @@ public class HudiController extends BaseController {
}
@GetMapping("read_compaction_plan")
public AmisResponse readCompactionPlan(
public AmisResponse<HudiCompactionPlan> readCompactionPlan(
@RequestParam(value = "flink_job_id", required = false) Long flinkJobId,
@RequestParam(value = "alias", required = false) String alias,
@RequestParam(value = "hdfs", required = false) String hdfs,
@RequestParam("instant") String instant
) throws Exception {
if (StrUtil.isNotBlank(hdfs)) {
return AmisResponse.responseDetail(hudiService.readCompactionPlanHdfs(hdfs, instant));
return AmisResponse.responseSuccess(hudiService.readCompactionPlanHdfs(hdfs, instant));
} else if (ObjectUtil.isNotNull(flinkJobId) && StrUtil.isNotBlank(alias)) {
return AmisResponse.responseDetail(hudiService.readCompactionPlan(flinkJobId, alias, instant));
return AmisResponse.responseSuccess(hudiService.readCompactionPlan(flinkJobId, alias, instant));
}
throw new Exception("Flink job id and alias or hdfs cannot be blank");
}

15
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/LogController.java
View File

@@ -1,9 +1,10 @@
package com.lanyuanxiaoyao.service.web.controller;
import cn.hutool.core.map.MapUtil;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.loki.LokiLogLine;
import com.lanyuanxiaoyao.service.forest.service.LokiService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisDetailResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,8 +38,8 @@ public class LogController extends BaseController {
}
@GetMapping("query_sync_log")
public AmisResponse querySyncLog(@RequestParam("flink_job_id") Long flinkJobId) {
return AmisResponse.responseDetail(lokiService.queryRange(MapUtil.<String, String>builder()
public AmisDetailResponse querySyncLog(@RequestParam("flink_job_id") Long flinkJobId) {
return AmisResponse.responseDetailData(lokiService.queryRange(MapUtil.<String, String>builder()
.put("app", "hudi-sync")
.put("run_type", "sync")
.put("flink_job_id", flinkJobId.toString())
@@ -50,8 +51,8 @@ public class LogController extends BaseController {
}
@GetMapping("query_compaction_log")
public AmisResponse queryCompactionLog(@RequestParam("flink_job_id") Long flinkJobId, @RequestParam("alias") String alias) {
return AmisResponse.responseDetail(lokiService.queryRange(MapUtil.<String, String>builder()
public AmisDetailResponse queryCompactionLog(@RequestParam("flink_job_id") Long flinkJobId, @RequestParam("alias") String alias) {
return AmisResponse.responseDetailData(lokiService.queryRange(MapUtil.<String, String>builder()
.put("app", "hudi-sync")
.put("run_type", "compaction")
.put("flink_job_id", flinkJobId.toString())
@@ -64,8 +65,8 @@ public class LogController extends BaseController {
}
@GetMapping("query_application_log")
public AmisResponse queryApplicationLog(@RequestParam("application_id") String applicationId) {
return AmisResponse.responseDetail(lokiService.queryRange(MapUtil.<String, String>builder()
public AmisDetailResponse queryApplicationLog(@RequestParam("application_id") String applicationId) {
return AmisResponse.responseDetailData(lokiService.queryRange(MapUtil.<String, String>builder()
.put("app", "hudi-sync")
.put("app_id", applicationId)
.build())

65
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/OverviewController.java
View File

@@ -9,6 +9,8 @@ import com.lanyuanxiaoyao.service.forest.service.InfoService;
import com.lanyuanxiaoyao.service.forest.service.QueueService;
import com.lanyuanxiaoyao.service.forest.service.ScheduleService;
import com.lanyuanxiaoyao.service.forest.service.YarnService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisMapResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.JobIdAndAliasVO;
@@ -35,7 +37,7 @@ import org.springframework.web.bind.annotation.RestController;
@RequestMapping("overview")
public class OverviewController extends BaseController {
private static final Logger logger = LoggerFactory.getLogger(OverviewController.class);
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
private final InfoService infoService;
private final YarnService yarnService;
private final QueueService queueService;
@@ -50,7 +52,7 @@ public class OverviewController extends BaseController {
}
@GetMapping("")
public AmisResponse overview() throws ExecutionException, InterruptedException {
public AmisMapResponse overview() throws ExecutionException, InterruptedException {
CompletableFuture<Long> tableCountFuture = CompletableFuture.supplyAsync(infoService::tableCount, ExecutorProvider.EXECUTORS);
CompletableFuture<Long> tableFocusCountFuture = CompletableFuture.supplyAsync(infoService::tableFocusCount, ExecutorProvider.EXECUTORS);
CompletableFuture<Long> hudiCountFuture = CompletableFuture.supplyAsync(infoService::hudiCount, ExecutorProvider.EXECUTORS);
@@ -65,66 +67,63 @@ public class OverviewController extends BaseController {
hiveCountFuture,
hiveFocusCountFuture
).get();
return AmisResponse.responseData()
.withData("table_count", tableCountFuture.get())
.withData("table_focus_count", tableFocusCountFuture.get())
.withData("hudi_count", hudiCountFuture.get())
.withData("hudi_focus_count", hudiFocusCountFuture.get())
.withData("hive_count", hiveCountFuture.get())
.withData("hive_focus_count", hiveFocusCountFuture.get());
return AmisResponse.responseMapData()
.setData("table_count", tableCountFuture.get())
.setData("table_focus_count", tableFocusCountFuture.get())
.setData("hudi_count", hudiCountFuture.get())
.setData("hudi_focus_count", hudiFocusCountFuture.get())
.setData("hive_count", hiveCountFuture.get())
.setData("hive_focus_count", hiveFocusCountFuture.get());
}
@GetMapping("yarn-job")
private AmisResponse yarnOverview(@RequestParam("cluster") String cluster, @RequestParam("search") String text) {
private AmisMapResponse 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 AmisResponse.responseData()
.withData("name", cluster)
.withData("total", applications.size())
.withData("running", applications.count(app -> StrUtil.equals(app.getState(), "RUNNING")))
.withData("running", applications.count(app -> StrUtil.equals(app.getState(), "RUNNING")))
.withData("scheduling", applications.count(app -> StrUtil.equals(app.getState(), "ACCEPTED")))
.withData("failure", applications.count(app -> StrUtil.equals(app.getState(), "FAILED")));
return AmisResponse.responseMapData()
.setData("name", cluster)
.setData("total", applications.size())
.setData("running", applications.count(app -> StrUtil.equals(app.getState(), "RUNNING")))
.setData("running", applications.count(app -> StrUtil.equals(app.getState(), "RUNNING")))
.setData("scheduling", applications.count(app -> StrUtil.equals(app.getState(), "ACCEPTED")))
.setData("failure", applications.count(app -> StrUtil.equals(app.getState(), "FAILED")));
}
@GetMapping("yarn-cluster")
public AmisResponse yarnClusterOverview(@RequestParam("cluster") String cluster, @RequestParam("queue") String queue) {
AmisResponse response = AmisResponse.responseData();
public AmisMapResponse yarnClusterOverview(@RequestParam("cluster") String cluster, @RequestParam("queue") String queue) {
AmisMapResponse response = AmisResponse.responseMapData();
YarnRootQueue root = yarnService.cluster(cluster);
response.withData("root", root);
response.setData("root", root);
if (StrUtil.isNotBlank(queue)) {
yarnService.queueList(cluster)
.select(q -> StrUtil.equals(q.getQueueName(), queue))
.getFirstOptional()
.ifPresent(yarnQueue -> response.withData("target", yarnQueue));
.ifPresent(yarnQueue -> response.setData("target", yarnQueue));
}
return response;
}
@GetMapping("queue")
public AmisResponse queueOverview(@RequestParam("queue") String queue) {
return AmisResponse.responseData()
.withData("size", queueService.size(queue));
public AmisMapResponse queueOverview(@RequestParam("queue") String queue) {
return AmisResponse.responseMapData("size", queueService.size(queue));
}
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
@GetMapping("version")
public AmisResponse versionOverview() throws ExecutionException, InterruptedException {
public AmisMapResponse 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).get();
return AmisResponse.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()));
return AmisResponse.responseMapData()
.setData("version", version)
.setData("unReceive", Maps.immutable.of("normal", unReceiveNormalTableCount.get(), "focus", unReceiveFocusCount.get()))
.setData("unSchedule", Maps.immutable.of("normal", unScheduledNormalTableCount.get(), "focus", unScheduledFocusTableCount.get()));
}
@GetMapping("version_detail")
public AmisResponse versionDetail(String target, String version) throws Exception {
public AmisCrudResponse versionDetail(String target, String version) throws Exception {
ImmutableList<JobIdAndAlias> jobIdAndAliases;
if (StrUtil.equals("unReceive_focus", target)) {
jobIdAndAliases = infoService.unReceiveVersionFocusTable(version);
@@ -141,7 +140,7 @@ public class OverviewController extends BaseController {
}
@GetMapping("schedule_jobs")
public AmisResponse scheduleJobs() {
public AmisCrudResponse scheduleJobs() {
return AmisResponse.responseCrudData(scheduleService.scheduleJobs());
}
}

6
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/PulsarController.java
View File

@@ -1,7 +1,7 @@
package com.lanyuanxiaoyao.service.web.controller;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.forest.service.PulsarService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.TopicVO;
import org.slf4j.Logger;
@@ -30,7 +30,7 @@ public class PulsarController extends BaseController {
}
@GetMapping("topic")
public AmisResponse topic(@RequestParam("pulsar_url") String pulsarUrl, @RequestParam("topic") String topic) {
return AmisResponse.responseDetail(new TopicVO(pulsarService.topic(pulsarService.name(pulsarUrl), topic)));
public AmisResponse<TopicVO> topic(@RequestParam("pulsar_url") String pulsarUrl, @RequestParam("topic") String topic) {
return AmisResponse.responseSuccess(new TopicVO(pulsarService.topic(pulsarService.name(pulsarUrl), topic)));
}
}

5
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/QueueController.java
View File

@@ -1,8 +1,9 @@
package com.lanyuanxiaoyao.service.web.controller;
import com.lanyuanxiaoyao.micro.service.entity.queue.QueueItem;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.forest.service.QueueService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,7 +31,7 @@ public class QueueController extends BaseController {
}
@GetMapping("all")
public AmisResponse all(@RequestParam("name") String name) {
public AmisCrudResponse all(@RequestParam("name") String name) {
return AmisResponse.responseCrudData(queueService.all(name).toSortedList(QueueItem::compareTo));
}
}

7
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/RunningController.java
View File

@@ -5,9 +5,10 @@ import com.eshore.odcp.hudi.connector.utils.NameHelper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.zookeeper.ZookeeperNode;
import com.lanyuanxiaoyao.service.forest.service.ZookeeperService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.ZookeeperNodeVO;
import org.slf4j.Logger;
@@ -38,7 +39,7 @@ public class RunningController extends BaseController {
}
@GetMapping("sync")
public AmisResponse sync() {
public AmisCrudResponse sync() {
return AmisResponse.responseCrudData(
zookeeperService.getChildren(NameHelper.ZK_SYNC_RUNNING_LOCK_PATH)
.asParallel(ExecutorProvider.EXECUTORS, 1)
@@ -50,7 +51,7 @@ public class RunningController extends BaseController {
}
@GetMapping("compaction")
public AmisResponse compaction() {
public AmisCrudResponse compaction() {
return AmisResponse.responseCrudData(
zookeeperService.getChildren(NameHelper.ZK_COMPACTION_RUNNING_LOCK_PATH)
.asParallel(ExecutorProvider.EXECUTORS, 1)

16
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/TableController.java
View File

@@ -9,12 +9,14 @@ import com.eshore.odcp.hudi.connector.entity.TableMeta;
import com.eshore.odcp.hudi.connector.utils.NameHelper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.PageResponse;
import com.lanyuanxiaoyao.service.configuration.entity.info.CompactionMetrics;
import com.lanyuanxiaoyao.service.configuration.entity.info.JobIdAndAlias;
import com.lanyuanxiaoyao.service.forest.service.InfoService;
import com.lanyuanxiaoyao.service.forest.service.ZookeeperService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisMapResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.CompactionMetricsVO;
import com.lanyuanxiaoyao.service.web.entity.FlinkJobVO;
@@ -142,7 +144,7 @@ public class TableController extends BaseController {
}
@GetMapping("list_metas")
public AmisResponse listTableMetas(@RequestParam("flink_job_id") Long flinkJobId) throws Exception {
public AmisCrudResponse listTableMetas(@RequestParam("flink_job_id") Long flinkJobId) throws Exception {
if (ObjectUtil.isNull(flinkJobId)) {
throw new Exception("flink job id is null");
}
@@ -150,20 +152,20 @@ public class TableController extends BaseController {
}
@GetMapping("detail")
public AmisResponse getMeta(@RequestParam("flink_job_id") Long flinkJobId, @RequestParam("alias") String alias) throws Exception {
public AmisMapResponse getMeta(@RequestParam("flink_job_id") Long flinkJobId, @RequestParam("alias") String alias) throws Exception {
if (ObjectUtil.isNull(flinkJobId) || ObjectUtil.isNull(alias)) {
throw new Exception("flink job id or alias is null");
}
CompletableFuture<FlinkJob> flinkJobFuture = CompletableFuture.supplyAsync(() -> infoService.flinkJobDetail(flinkJobId), ExecutorProvider.EXECUTORS);
CompletableFuture<TableMeta> tableMetaFuture = CompletableFuture.supplyAsync(() -> infoService.tableMetaDetail(flinkJobId, alias), ExecutorProvider.EXECUTORS);
CompletableFuture.allOf(flinkJobFuture, tableMetaFuture).get();
return AmisResponse.responseData()
.withData("flinkJob", new FlinkJobVO(flinkJobFuture.get()))
.withData("tableMeta", tableMetaFuture.get());
return AmisResponse.responseMapData()
.setData("flinkJob", new FlinkJobVO(flinkJobFuture.get()))
.setData("tableMeta", tableMetaFuture.get());
}
@GetMapping("list_compaction_metrics")
public AmisResponse listCompactionMetrics(
public AmisCrudResponse listCompactionMetrics(
@RequestParam(value = "page", defaultValue = "1") Integer page,
@RequestParam(value = "count", defaultValue = "10") Integer count,
@RequestParam(value = "order", required = false) String order,

9
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/VersionController.java
View File

@@ -5,10 +5,11 @@ import cn.hutool.core.util.StrUtil;
import com.eshore.odcp.hudi.connector.entity.FlinkJob;
import com.eshore.odcp.hudi.connector.entity.TableMeta;
import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.PageResponse;
import com.lanyuanxiaoyao.service.configuration.entity.info.VersionUpdated;
import com.lanyuanxiaoyao.service.forest.service.InfoService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisMapResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.VersionUpdateVO;
import java.util.List;
@@ -43,7 +44,7 @@ public class VersionController extends BaseController {
}
@GetMapping("list")
public AmisResponse list(
public AmisMapResponse list(
@RequestParam(value = "page", defaultValue = "1") Integer page,
@RequestParam(value = "count", defaultValue = "10") Integer count,
@RequestParam(value = "order", required = false) String order,
@@ -87,7 +88,7 @@ public class VersionController extends BaseController {
.toList()
.toImmutable();
return AmisResponse.responseCrudData(vos, total)
.withData("scheduled", pageResponse.getMetadata().get("scheduled"))
.withData("unScheduled", pageResponse.getMetadata().get("unScheduled"));
.setData("scheduled", pageResponse.getMetadata().get("scheduled"))
.setData("unScheduled", pageResponse.getMetadata().get("unScheduled"));
}
}

35
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/YarnController.java
View File

@@ -1,21 +1,22 @@
package com.lanyuanxiaoyao.service.web.controller;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.flink.FlinkVertex;
import com.lanyuanxiaoyao.service.configuration.entity.flink.FlinkVertexOverview;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnApplication;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnQueue;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnRootQueue;
import com.lanyuanxiaoyao.service.configuration.utils.ComparatorUtil;
import com.lanyuanxiaoyao.service.forest.service.FlinkService;
import com.lanyuanxiaoyao.service.forest.service.YarnService;
import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisMapResponse;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.web.controller.base.BaseController;
import com.lanyuanxiaoyao.service.web.entity.YarnApplicationVO;
import com.lanyuanxiaoyao.service.web.entity.YarnClusterVO;
import com.lanyuanxiaoyao.service.configuration.utils.ComparatorUtil;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
@@ -57,7 +58,7 @@ public class YarnController extends BaseController {
}
@GetMapping("job_list")
public AmisResponse jobList(
public AmisMapResponse jobList(
@RequestParam("clusters") List<String> clusters,
@RequestParam(value = "page", defaultValue = "1") Integer page,
@RequestParam(value = "count", defaultValue = "10") Integer count,
@@ -120,12 +121,12 @@ public class YarnController extends BaseController {
.toList()
.toImmutable();
return AmisResponse.responseCrudData(result, applications.size())
.withData("running", running)
.withData("unRunning", unRunning);
.setData("running", running)
.setData("unRunning", unRunning);
}
@GetMapping("job_current")
public AmisResponse jobCurrent(@RequestParam("clusters") List<String> clusters, @RequestParam("name") String name) {
public AmisMapResponse jobCurrent(@RequestParam("clusters") List<String> clusters, @RequestParam("name") String name) {
Optional<YarnApplication> currentApp = Lists.immutable.ofAll(clusters)
.asParallel(ExecutorProvider.EXECUTORS, 1)
.flatCollect(cluster -> yarnService.jobListEquals(cluster, name))
@@ -133,16 +134,16 @@ public class YarnController extends BaseController {
.toSortedList(ComparatorUtil.longComparator("startedTime", ComparatorUtil.DESC, SORT_MAP))
.getFirstOptional();
if (currentApp.isPresent()) {
return AmisResponse.responseData()
.withData("hasCurrent", true)
.withData("current", currentApp.get());
return AmisResponse.responseMapData()
.setData("hasCurrent", true)
.setData("current", currentApp.get());
} else {
return AmisResponse.responseData().withData("hasCurrent", false);
return AmisResponse.responseMapData("hasCurrent", false);
}
}
@GetMapping("queue_list")
public AmisResponse queueList(@RequestParam("clusters") List<String> clusters, @RequestParam(value = "names", defaultValue = "") String names) {
public AmisCrudResponse queueList(@RequestParam("clusters") List<String> clusters, @RequestParam(value = "names", defaultValue = "") String names) {
boolean isFilterNames = StrUtil.isNotBlank(names);
ImmutableList<String> filterNames = Lists.immutable.of(names.split(","));
ImmutableList<YarnClusterVO> results = Lists.immutable.ofAll(clusters)
@@ -158,23 +159,23 @@ public class YarnController extends BaseController {
}
@GetMapping("queue_names")
public AmisResponse queueNames(@RequestParam("clusters") List<String> clusters) {
public AmisMapResponse queueNames(@RequestParam("clusters") List<String> clusters) {
ImmutableList<String> names = Lists.immutable.ofAll(clusters)
.asParallel(ExecutorProvider.EXECUTORS, 1)
.flatCollect(yarnService::queueList)
.collect(YarnQueue::getQueueName)
.toList()
.toImmutable();
return AmisResponse.responseData(MapUtil.of("queueNames", names));
return AmisResponse.responseMapData("queueNames", names);
}
@GetMapping("clusters")
public AmisResponse clusters(@RequestParam("clusters") List<String> clusters) {
ImmutableList<YarnRootQueue> roots = (ImmutableList<YarnRootQueue>) Lists.immutable.ofAll(clusters)
public AmisMapResponse clusters(@RequestParam("clusters") List<String> clusters) {
ImmutableList<YarnRootQueue> roots = Lists.immutable.ofAll(clusters)
.asParallel(ExecutorProvider.EXECUTORS, 1)
.collect(yarnService::cluster)
.toList()
.toImmutable();
return AmisResponse.responseData(MapUtil.of("cluster", clusters));
return AmisResponse.responseMapData("cluster", clusters);
}
}

30
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/base/AmisCrudResponse.java Normal file
View File

@@ -0,0 +1,30 @@
package com.lanyuanxiaoyao.service.web.controller.base;
/**
* Crud 响应
*
* @author lanyuanxiaoyao
* @date 2023-07-06
*/
public class AmisCrudResponse extends AmisMapResponse {
public void setData(Iterable<?> list) {
getData().put("item", list);
}
public void setTotal(Long total) {
getData().put("total", total);
}
public void setTotal(Integer total) {
setTotal(total.longValue());
}
public void setData(Iterable<?> list, Long total) {
setData(list);
setTotal(total);
}
public void setData(Iterable<?> list, Integer total) {
setData(list, total.longValue());
}
}

13
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/base/AmisDetailResponse.java Normal file
View File

@@ -0,0 +1,13 @@
package com.lanyuanxiaoyao.service.web.controller.base;
/**
* Crud 响应
*
* @author lanyuanxiaoyao
* @date 2023-07-06
*/
public class AmisDetailResponse extends AmisMapResponse {
public void setDetail(Object detail) {
getData().put("detail", detail);
}
}

21
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/base/AmisMapResponse.java Normal file
View File

@@ -0,0 +1,21 @@
package com.lanyuanxiaoyao.service.web.controller.base;
import java.util.HashMap;
import java.util.Map;
/**
* Map 响应
*
* @author lanyuanxiaoyao
* @date 2023-07-06
*/
public class AmisMapResponse extends AmisResponse<Map<String, Object>> {
public AmisMapResponse() {
setData(new HashMap<>());
}
public AmisMapResponse setData(String key, Object value) {
getData().put(key, value);
return this;
}
}

208
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/base/AmisResponse.java
View File

@@ -1,7 +1,5 @@
package com.lanyuanxiaoyao.service.web.controller.base;
import cn.hutool.core.map.MapUtil;
import java.util.HashMap;
import java.util.Map;
/**
@@ -10,19 +8,102 @@ import java.util.Map;
* @author lanyuanxiaoyao
* @date 2022-09-21
*/
public class AmisResponse {
public class AmisResponse<T> {
private static final int SUCCESS_STATUS = 0;
private static final int ERROR_STATUS = 500;
private static final String SUCCESS_MESSAGE = "OK";
private static final String ERROR_MESSAGE = "ERROR";
private Integer status;
private String message;
private Map<String, Object> data;
private T data;
public AmisResponse(Builder builder) {
this.status = builder.status;
this.message = builder.message;
this.data = builder.data;
public static AmisResponse<Object> responseError() {
AmisResponse<Object> response = new AmisResponse<>();
response.setStatus(ERROR_STATUS);
response.setMessage(ERROR_MESSAGE);
return response;
}
public static Builder builder() {
return new Builder();
public static AmisResponse<Object> responseError(String message) {
AmisResponse<Object> response = new AmisResponse<>();
response.setStatus(ERROR_STATUS);
response.setMessage(message);
return response;
}
public static AmisResponse<Object> responseSuccess() {
AmisResponse<Object> response = new AmisResponse<>();
response.setStatus(SUCCESS_STATUS);
response.setMessage(SUCCESS_MESSAGE);
return response;
}
public static AmisResponse<Object> responseSuccess(String message) {
AmisResponse<Object> response = new AmisResponse<>();
response.setStatus(SUCCESS_STATUS);
response.setMessage(message);
return response;
}
public static <E> AmisResponse<E> responseSuccess(String message, E data) {
AmisResponse<E> response = new AmisResponse<>();
response.setStatus(SUCCESS_STATUS);
response.setMessage(message);
response.setData(data);
return response;
}
public static <E> AmisResponse<E> responseSuccess(E data) {
AmisResponse<E> response = new AmisResponse<>();
response.setStatus(SUCCESS_STATUS);
response.setMessage(SUCCESS_MESSAGE);
response.setData(data);
return response;
}
public static AmisMapResponse responseMapData() {
AmisMapResponse response = new AmisMapResponse();
response.setStatus(SUCCESS_STATUS);
response.setMessage(SUCCESS_MESSAGE);
return response;
}
public static AmisMapResponse responseMapData(Map<String, Object> data) {
AmisMapResponse response = responseMapData();
response.setData(data);
return response;
}
public static AmisMapResponse responseMapData(String key, Object value) {
AmisMapResponse response = responseMapData();
response.setData(key, value);
return response;
}
public static AmisCrudResponse responseCrudData(Iterable<?> data) {
AmisCrudResponse response = new AmisCrudResponse();
response.setStatus(SUCCESS_STATUS);
response.setMessage(SUCCESS_MESSAGE);
response.setData(data);
return response;
}
public static AmisCrudResponse responseCrudData(Iterable<?> data, Integer total) {
AmisCrudResponse response = responseCrudData(data);
response.setTotal(total);
return response;
}
public static AmisCrudResponse responseCrudData(Iterable<?> data, Long total) {
AmisCrudResponse response = responseCrudData(data);
response.setTotal(total);
return response;
}
public static AmisDetailResponse responseDetailData(Object detail) {
AmisDetailResponse response = new AmisDetailResponse();
response.setDetail(detail);
return response;
}
public Integer getStatus() {
@@ -41,19 +122,14 @@ public class AmisResponse {
this.message = message;
}
public Map<String, Object> getData() {
public T getData() {
return data;
}
public void setData(Map<String, Object> data) {
public void setData(T data) {
this.data = data;
}
public AmisResponse withData(String key, Object value) {
data.put(key, value);
return this;
}
@Override
public String toString() {
return "AmisResponse{" +
@@ -62,102 +138,4 @@ public class AmisResponse {
", data=" + data +
'}';
}
public static final class Builder {
private Integer status = 0;
private String message = "";
private Map<String, Object> data = null;
private Builder() {
}
public Builder status(Integer status) {
this.status = status;
return this;
}
public Builder message(String message) {
this.message = message;
return this;
}
public Builder data(Map<String, Object> data) {
this.data = data;
return this;
}
public AmisResponse build() {
return new AmisResponse(this);
}
}
private static final int SUCCESS_STATUS = 0;
private static final int ERROR_STATUS = 500;
private static final String SUCCESS_MESSAGE = "OK";
private static final String ERROR_MESSAGE = "ERROR";
public static AmisResponse responseError() {
return AmisResponse.builder()
.status(ERROR_STATUS)
.message(ERROR_MESSAGE)
.build();
}
public static AmisResponse responseError(String message) {
return AmisResponse.builder()
.status(ERROR_STATUS)
.message(message)
.build();
}
public static AmisResponse responseData() {
return AmisResponse.builder()
.status(SUCCESS_STATUS)
.message(SUCCESS_MESSAGE)
.data(new HashMap<>())
.build();
}
public static AmisResponse responseData(Map<String, Object> data) {
return AmisResponse.builder()
.status(SUCCESS_STATUS)
.message(SUCCESS_MESSAGE)
.data(data)
.build();
}
public static AmisResponse responseDetail(Object detail) {
return AmisResponse.builder()
.status(SUCCESS_STATUS)
.message(SUCCESS_MESSAGE)
.data(MapUtil.<String, Object>builder()
.put("detail", detail)
.build())
.build();
}
public static AmisResponse responseCrudData(Iterable<?> data) {
return AmisResponse.builder()
.status(SUCCESS_STATUS)
.message(SUCCESS_MESSAGE)
.data(MapUtil.<String, Object>builder()
.put("items", data)
.build())
.build();
}
public static AmisResponse responseCrudData(Iterable<?> data, Integer total) {
return responseCrudData(data, Integer.toUnsignedLong(total));
}
public static AmisResponse responseCrudData(Iterable<?> data, Long total) {
return AmisResponse.builder()
.status(SUCCESS_STATUS)
.message(SUCCESS_MESSAGE)
.data(MapUtil.<String, Object>builder()
.put("items", data)
.put("total", total)
.build())
.build();
}
}

4
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/base/BaseController.java
View File

@@ -1,13 +1,9 @@
package com.lanyuanxiaoyao.service.web.controller.base;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.configuration.Constants;
import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.exception.TableNotFoundException;
import com.lanyuanxiaoyao.service.forest.service.InfoService;
import java.util.HashMap;
import java.util.Map;
import org.eclipse.collections.api.factory.Maps;
import org.eclipse.collections.api.map.MutableMap;

2
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/base/ErrorResponseAdvice.java
View File

@@ -1,8 +1,6 @@
package com.lanyuanxiaoyao.service.web.controller.base;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.http.HttpStatus;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestControllerAdvice;

46
web/components/common.js
View File

@@ -456,15 +456,15 @@ function simpleYarnDialog(cluster, title, filterField) {
title: 'Flink 基本信息',
column: 4,
items: [
{label: 'Flink 版本', content: '${detail.flinkVersion}'},
{label: 'Flink 小版本', content: '${detail.flinkCommit}', span: 3},
{label: '运行中', content: '${detail.jobsRunning}'},
{label: '已结束', content: '${detail.jobsFinished}'},
{label: '已失败', content: '${detail.jobsFailed}'},
{label: '被取消', content: '${detail.jobsCanceled}'},
{label: 'Flink 版本', content: '${flinkVersion}'},
{label: 'Flink 小版本', content: '${flinkCommit}', span: 3},
{label: '运行中', content: '${jobsRunning}'},
{label: '已结束', content: '${jobsFinished}'},
{label: '已失败', content: '${jobsFailed}'},
{label: '被取消', content: '${jobsCanceled}'},
{
label: 'Slot (可用/总数)',
content: '${detail.slotsAvailable}/${detail.slotsTotal}',
content: '${slotsAvailable}/${slotsTotal}',
span: 4
},
]
@@ -816,7 +816,7 @@ function timelineColumns() {
},
adaptor: (payload, response) => {
return {
items: (payload['data']['detail']['operations'] ? payload['data']['detail']['operations'] : [])
items: (payload['data']['operations'] ? payload['data']['operations'] : [])
.map(operation => {
if (operation['deltaFilePaths']) {
operation.deltaFilePaths = operation.deltaFilePaths
@@ -1132,8 +1132,8 @@ function tableMetaDialog() {
type: 'property',
title: '基本信息',
items: [
{label: 'Topic', content: copyField('detail.name'), span: 2},
{label: '最末位移', content: copyField('detail.lastMessageId')},
{label: 'Topic', content: copyField('name'), span: 2},
{label: '最末位移', content: copyField('lastMessageId')},
],
},
{type: 'divider'},
@@ -1142,40 +1142,40 @@ function tableMetaDialog() {
title: '指标信息',
column: 4,
items: [
{label: '入队列消息速率', content: '${detail.state.messageRateIn}'},
{label: '入队列消息速率', content: '${state.messageRateIn}'},
{
label: '出队列消息速率',
content: '${detail.state.messageRateOut}'
content: '${state.messageRateOut}'
},
{
label: '入队列消息吞吐量',
content: '${detail.state.messageThroughputIn}'
content: '${state.messageThroughputIn}'
},
{
label: '出队列消息吞吐量',
content: '${detail.state.messageThroughputOut}'
content: '${state.messageThroughputOut}'
},
{
label: '入队列消息数量',
content: '${detail.state.messageInCounter}'
content: '${state.messageInCounter}'
},
{
label: '出队列消息数量',
content: '${detail.state.messageOutCounter}'
content: '${state.messageOutCounter}'
},
{
label: '入队列消息字节数',
content: '${detail.state.byteInCounter}'
content: '${state.byteInCounter}'
},
{
label: '出队列消息字节数',
content: '${detail.state.byteOutCounter}'
content: '${state.byteOutCounter}'
},
{label: '存储消息大小', content: '${detail.state. storageSize}'},
{label: '积压消息大小', content: '${detail.state.backlogSize}'},
{label: '存储消息大小', content: '${state.storageSize}'},
{label: '积压消息大小', content: '${state.backlogSize}'},
{
label: '平均消息大小',
content: '${detail.state.averageMessageSize}'
content: '${state.averageMessageSize}'
},
],
},
@@ -1183,7 +1183,7 @@ function tableMetaDialog() {
{
type: 'table',
title: '消费者们',
source: '${detail.subscriptionStateVOS}',
source: '${subscriptionStateVOS}',
itemAction: {
type: 'action',
actionType: 'dialog',
@@ -1315,7 +1315,7 @@ function tableMetaDialog() {
{
type: 'table',
title: '生产者们',
source: '${detail.state.publishers}',
source: '${state.publishers}',
itemAction: {
type: 'action',
actionType: 'dialog',