This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new daed18b [INLONG-2957][Manager] Optimize the cluster management
interface (#2959)
daed18b is described below
commit daed18b359a337cc2a6ed68190fb4f18b48bb988
Author: healchow <[email protected]>
AuthorDate: Mon Mar 7 14:22:17 2022 +0800
[INLONG-2957][Manager] Optimize the cluster management interface (#2959)
---
.../apache/inlong/agent/core/HeartbeatManager.java | 26 +++++----
.../manager/web/controller/ClusterController.java | 66 ++++++++++++++++++----
.../{ => openapi}/ClusterController.java | 44 ++++++++-------
3 files changed, 93 insertions(+), 43 deletions(-)
diff --git
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
index 783168f..56fd8e7 100644
---
a/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
+++
b/inlong-agent/agent-core/src/main/java/org/apache/inlong/agent/core/HeartbeatManager.java
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
@@ -17,7 +17,6 @@
package org.apache.inlong.agent.core;
-import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.apache.inlong.agent.common.AbstractDaemon;
import org.apache.inlong.agent.conf.AgentConfiguration;
@@ -34,6 +33,8 @@ import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
import static
org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_REPORTSNAPSHOT_HTTP_PATH;
import static
org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_HOST;
@@ -43,7 +44,7 @@ import static
org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MA
import static
org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_VIP_HTTP_PREFIX_PATH;
import static
org.apache.inlong.agent.core.task.TaskPositionManager.DEFAULT_FLUSH_TIMEOUT;
-public class HeartbeatManager extends AbstractDaemon {
+public class HeartbeatManager extends AbstractDaemon {
private static final Logger LOGGER =
LoggerFactory.getLogger(HeartbeatManager.class);
@@ -53,6 +54,7 @@ public class HeartbeatManager extends AbstractDaemon {
private final HttpManager httpManager;
private final String baseManagerUrl;
private final String reportSnapshotUrl;
+ private final Pattern numberPattern = Pattern.compile("^[-+]?[\\d]*$");
/**
* Init heartbeat manager.
@@ -68,25 +70,27 @@ public class HeartbeatManager extends AbstractDaemon {
/**
* fetch heartbeat of job
- * @return
*/
private TaskSnapshotRequest getHeartBeat() {
Map<String, JobWrapper> jobWrapperMap = jobmanager.getJobs();
-
List<TaskSnapshotMessage> taskSnapshotMessageList = new ArrayList<>();
TaskSnapshotRequest taskSnapshotRequest = new TaskSnapshotRequest();
Date date = new Date(System.currentTimeMillis());
-
- for (Map.Entry<String, JobWrapper> entry:jobWrapperMap.entrySet()) {
+ for (Map.Entry<String, JobWrapper> entry : jobWrapperMap.entrySet()) {
if (StringUtils.isBlank(entry.getKey()) || entry.getValue() ==
null) {
- LOGGER.info(" key : {}, value : {} exits
null",entry.getKey(),entry.getValue());
+ LOGGER.info("key: {} or value: {} is null", entry.getKey(),
entry.getValue());
continue;
}
String offset = entry.getValue().getSnapshot();
String jobId = entry.getKey();
TaskSnapshotMessage snapshotMessage = new TaskSnapshotMessage();
snapshotMessage.setSnapshot(offset);
+
+ // TODO Need to make sure the jobId is an Integer
+ if (!numberPattern.matcher(jobId).matches()) {
+ continue;
+ }
snapshotMessage.setJobId(Integer.valueOf(jobId));
taskSnapshotMessageList.add(snapshotMessage);
}
@@ -100,7 +104,7 @@ public class HeartbeatManager extends AbstractDaemon {
/**
* build base url for manager according to config
*
- * @example - http://127.0.0.1:8080/api/inlong/manager/openapi
+ * example - http://127.0.0.1:8080/api/inlong/manager/openapi
*/
private String buildBaseUrl() {
return "http://"; + conf.get(AGENT_MANAGER_VIP_HTTP_HOST)
@@ -123,8 +127,8 @@ public class HeartbeatManager extends AbstractDaemon {
while (isRunnable()) {
try {
TaskSnapshotRequest taskSnapshotRequest = getHeartBeat();
-
httpManager.doSentPost(reportSnapshotUrl,taskSnapshotRequest);
- LOGGER.info(" {} report to manager",taskSnapshotRequest);
+ httpManager.doSentPost(reportSnapshotUrl,
taskSnapshotRequest);
+ LOGGER.info(" {} report to manager", taskSnapshotRequest);
TimeUnit.SECONDS.sleep(DEFAULT_FLUSH_TIMEOUT);
} catch (Exception ex) {
LOGGER.error("error caught", ex);
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
index 6d36a8f..5ec249a 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
@@ -28,11 +28,13 @@ import
org.apache.inlong.manager.common.pojo.cluster.ClusterRequest;
import org.apache.inlong.manager.common.pojo.cluster.DataProxyClusterInfo;
import
org.apache.inlong.manager.common.pojo.cluster.DataProxyClusterPageRequest;
import org.apache.inlong.manager.common.util.LoginUserUtils;
-import org.apache.inlong.manager.service.core.ThirdPartyClusterService;
import org.apache.inlong.manager.service.core.DataProxyClusterService;
+import org.apache.inlong.manager.service.core.ThirdPartyClusterService;
import org.apache.inlong.manager.service.core.operationlog.OperationLog;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
@@ -41,7 +43,7 @@ import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
- * Various cluster control layers
+ * Cluster controller
*/
@RestController
@RequestMapping("/cluster")
@@ -53,40 +55,82 @@ public class ClusterController {
@Autowired
private DataProxyClusterService dataProxyClusterService;
- @RequestMapping(value = "/thirdparty/list", method = RequestMethod.GET)
- @ApiOperation(value = "Query the list of general clusters based on
conditions")
- public Response<List<ClusterInfo>> list(ClusterRequest request) {
+ @RequestMapping(value = "/save", method = RequestMethod.GET)
+ @ApiOperation(value = "Save cluster info")
+ @OperationLog(operation = OperationType.CREATE)
+ public Response<Integer> save(@RequestBody ClusterInfo clusterInfo) {
+ String currentUser = LoginUserUtils.getLoginUserDetail().getUserName();
+ return Response.success(thirdPartyClusterService.save(clusterInfo,
currentUser));
+ }
+
+ @GetMapping(value = "/get/{id}")
+ @ApiOperation(value = "Get cluster info by id")
+ @ApiImplicitParam(name = "id", value = "common cluster ID", dataTypeClass
= Integer.class, required = true)
+ public Response<ClusterInfo> get(@PathVariable Integer id) {
+ return Response.success(thirdPartyClusterService.get(id));
+ }
+
+ @PostMapping(value = "/list")
+ @ApiOperation(value = "List clusters by condition")
+ public Response<List<ClusterInfo>> list(@RequestBody ClusterRequest
request) {
return Response.success(thirdPartyClusterService.list(request));
}
- @RequestMapping(value = "/thirdparty/save", method = RequestMethod.GET)
+ @PostMapping(value = "/update")
+ @OperationLog(operation = OperationType.UPDATE)
+ @ApiOperation(value = "Update cluster info")
+ public Response<Boolean> update(@RequestBody ClusterInfo clusterInfo) {
+ String username = LoginUserUtils.getLoginUserDetail().getUserName();
+ return Response.success(thirdPartyClusterService.update(clusterInfo,
username));
+ }
+
+ @RequestMapping(value = "/delete/{id}", method = {RequestMethod.POST,
RequestMethod.DELETE})
+ @ApiOperation(value = "Delete cluster info by id")
+ @OperationLog(operation = OperationType.DELETE)
+ @ApiImplicitParam(name = "id", value = "Cluster ID", dataTypeClass =
Integer.class, required = true)
+ public Response<Boolean> delete(@PathVariable Integer id) {
+ return Response.success(thirdPartyClusterService.delete(id,
LoginUserUtils.getLoginUserDetail().getUserName()));
+ }
+
+ @Deprecated
+ @PostMapping(value = "/thirdparty/save")
@ApiOperation(value = "Add a cluster info")
@OperationLog(operation = OperationType.CREATE)
- public Response<Integer> saveCluster(@RequestBody ClusterInfo clusterInfo)
{
+ public Response<Integer> saveClusterV1(@RequestBody ClusterInfo
clusterInfo) {
String currentUser = LoginUserUtils.getLoginUserDetail().getUserName();
return Response.success(thirdPartyClusterService.save(clusterInfo,
currentUser));
}
- @RequestMapping(value = "/thirdparty/get/{id}")
+ @Deprecated
+ @GetMapping(value = "/thirdparty/get/{id}")
@ApiOperation(value = "Query third party cluster information of the
common")
@ApiImplicitParam(name = "id", value = "common cluster ID", dataTypeClass
= Integer.class, required = true)
- public Response<ClusterInfo> getCluster(@PathVariable Integer id) {
+ public Response<ClusterInfo> getClusterV1(@PathVariable Integer id) {
return Response.success(thirdPartyClusterService.get(id));
}
+ @Deprecated
+ @PostMapping(value = "/thirdparty/list")
+ @ApiOperation(value = "Query the list of general clusters based on
conditions")
+ public Response<List<ClusterInfo>> listV1(@RequestBody ClusterRequest
request) {
+ return Response.success(thirdPartyClusterService.list(request));
+ }
+
+ @Deprecated
@RequestMapping(value = "/thirdparty/update", method = RequestMethod.POST)
@OperationLog(operation = OperationType.UPDATE)
@ApiOperation(value = "Modify third party cluster information of the
common")
- public Response<Boolean> updateCluster(@RequestBody ClusterInfo
clusterInfo) {
+ public Response<Boolean> updateClusterV1(@RequestBody ClusterInfo
clusterInfo) {
String username = LoginUserUtils.getLoginUserDetail().getUserName();
return Response.success(thirdPartyClusterService.update(clusterInfo,
username));
}
+ @Deprecated
@RequestMapping(value = "/thirdparty/delete/{id}", method =
RequestMethod.DELETE)
@ApiOperation(value = "Delete third party cluster information")
@OperationLog(operation = OperationType.DELETE)
@ApiImplicitParam(name = "id", value = "DataProxy cluster id",
dataTypeClass = Integer.class, required = true)
- public Response<Boolean> delete(@PathVariable Integer id) {
+ public Response<Boolean> deleteV1(@PathVariable Integer id) {
return Response.success(thirdPartyClusterService.delete(id,
LoginUserUtils.getLoginUserDetail().getUserName()));
}
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/ClusterController.java
similarity index 81%
copy from
inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
copy to
inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/ClusterController.java
index 6d36a8f..d95c1f4 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/ClusterController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/ClusterController.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.web.controller;
+package org.apache.inlong.manager.web.controller.openapi;
import com.github.pagehelper.PageInfo;
import io.swagger.annotations.Api;
@@ -28,11 +28,13 @@ import
org.apache.inlong.manager.common.pojo.cluster.ClusterRequest;
import org.apache.inlong.manager.common.pojo.cluster.DataProxyClusterInfo;
import
org.apache.inlong.manager.common.pojo.cluster.DataProxyClusterPageRequest;
import org.apache.inlong.manager.common.util.LoginUserUtils;
-import org.apache.inlong.manager.service.core.ThirdPartyClusterService;
import org.apache.inlong.manager.service.core.DataProxyClusterService;
+import org.apache.inlong.manager.service.core.ThirdPartyClusterService;
import org.apache.inlong.manager.service.core.operationlog.OperationLog;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
@@ -41,7 +43,7 @@ import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
- * Various cluster control layers
+ * Cluster controller
*/
@RestController
@RequestMapping("/cluster")
@@ -53,39 +55,39 @@ public class ClusterController {
@Autowired
private DataProxyClusterService dataProxyClusterService;
- @RequestMapping(value = "/thirdparty/list", method = RequestMethod.GET)
- @ApiOperation(value = "Query the list of general clusters based on
conditions")
- public Response<List<ClusterInfo>> list(ClusterRequest request) {
- return Response.success(thirdPartyClusterService.list(request));
- }
-
- @RequestMapping(value = "/thirdparty/save", method = RequestMethod.GET)
- @ApiOperation(value = "Add a cluster info")
+ @RequestMapping(value = "/save", method = RequestMethod.GET)
+ @ApiOperation(value = "Save cluster info")
@OperationLog(operation = OperationType.CREATE)
- public Response<Integer> saveCluster(@RequestBody ClusterInfo clusterInfo)
{
+ public Response<Integer> save(@RequestBody ClusterInfo clusterInfo) {
String currentUser = LoginUserUtils.getLoginUserDetail().getUserName();
return Response.success(thirdPartyClusterService.save(clusterInfo,
currentUser));
}
- @RequestMapping(value = "/thirdparty/get/{id}")
- @ApiOperation(value = "Query third party cluster information of the
common")
+ @GetMapping(value = "/get/{id}")
+ @ApiOperation(value = "Get cluster info by id")
@ApiImplicitParam(name = "id", value = "common cluster ID", dataTypeClass
= Integer.class, required = true)
- public Response<ClusterInfo> getCluster(@PathVariable Integer id) {
+ public Response<ClusterInfo> get(@PathVariable Integer id) {
return Response.success(thirdPartyClusterService.get(id));
}
- @RequestMapping(value = "/thirdparty/update", method = RequestMethod.POST)
+ @PostMapping(value = "/list")
+ @ApiOperation(value = "List clusters by condition")
+ public Response<List<ClusterInfo>> list(@RequestBody ClusterRequest
request) {
+ return Response.success(thirdPartyClusterService.list(request));
+ }
+
+ @PostMapping(value = "/update")
@OperationLog(operation = OperationType.UPDATE)
- @ApiOperation(value = "Modify third party cluster information of the
common")
- public Response<Boolean> updateCluster(@RequestBody ClusterInfo
clusterInfo) {
+ @ApiOperation(value = "Update cluster info")
+ public Response<Boolean> update(@RequestBody ClusterInfo clusterInfo) {
String username = LoginUserUtils.getLoginUserDetail().getUserName();
return Response.success(thirdPartyClusterService.update(clusterInfo,
username));
}
- @RequestMapping(value = "/thirdparty/delete/{id}", method =
RequestMethod.DELETE)
- @ApiOperation(value = "Delete third party cluster information")
+ @RequestMapping(value = "/delete/{id}", method = {RequestMethod.POST,
RequestMethod.DELETE})
+ @ApiOperation(value = "Delete cluster info by id")
@OperationLog(operation = OperationType.DELETE)
- @ApiImplicitParam(name = "id", value = "DataProxy cluster id",
dataTypeClass = Integer.class, required = true)
+ @ApiImplicitParam(name = "id", value = "Cluster ID", dataTypeClass =
Integer.class, required = true)
public Response<Boolean> delete(@PathVariable Integer id) {
return Response.success(thirdPartyClusterService.delete(id,
LoginUserUtils.getLoginUserDetail().getUserName()));
}
