This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 76c762d96 [INLONG-7730][Manager] Support node management for Redis
(#7741)
76c762d96 is described below
commit 76c762d9654d088cbb14456a1ea0a2eb1334bab0
Author: feat <[email protected]>
AuthorDate: Mon Apr 3 18:10:37 2023 +0800
[INLONG-7730][Manager] Support node management for Redis (#7741)
---
.../manager/pojo/node/redis/RedisDataNodeDTO.java | 37 ++++++++++++++++++++++
.../manager/pojo/node/redis/RedisDataNodeInfo.java | 37 ++++++++++++++++++++++
.../pojo/node/redis/RedisDataNodeRequest.java | 36 +++++++++++++++++++++
.../inlong/manager/pojo/sink/redis/RedisSink.java | 2 +-
.../manager/pojo/sink/redis/RedisSinkDTO.java | 2 +-
.../manager/pojo/sink/redis/RedisSinkRequest.java | 2 +-
.../manager/pojo/sort/util/LoadNodeUtils.java | 4 +--
.../service/node/redis/RedisDataNodeOperator.java | 35 ++++++++++++++++++++
.../service/sink/redis/RedisSinkOperator.java | 2 +-
.../sort/protocol/node/load/RedisLoadNode.java | 8 ++---
10 files changed, 155 insertions(+), 10 deletions(-)
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeDTO.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeDTO.java
index 1c6b64cd8..d27528ae7 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeDTO.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeDTO.java
@@ -18,6 +18,8 @@
package org.apache.inlong.manager.pojo.node.redis;
import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@@ -35,11 +37,46 @@ import javax.validation.constraints.NotNull;
@Data
@Builder
@NoArgsConstructor
+@AllArgsConstructor
@ApiModel("Redis data node info")
public class RedisDataNodeDTO {
private static final Logger LOGGER =
LoggerFactory.getLogger(RedisDataNodeDTO.class);
+ /**
+ * Redis cluster mode
+ */
+ @ApiModelProperty(value = "Redis cluster mode")
+ private String clusterMode;
+
+ /**
+ * Redis host
+ */
+ @ApiModelProperty(value = "Redis host")
+ private String host;
+
+ /**
+ * Redis port
+ */
+ @ApiModelProperty(value = "Redis port")
+ private Integer port;
+ /**
+ * Redis sentinel master name
+ */
+ @ApiModelProperty(value = "Redis sentinel master name")
+ private String masterName;
+
+ /**
+ * Redis sentinel info
+ */
+ @ApiModelProperty(value = "Redis sentinel info")
+ private String sentinelsInfo;
+
+ /**
+ * Redis cluster nodes
+ */
+ @ApiModelProperty(value = "Redis cluster nodes")
+ private String clusterNodes;
/**
* Get the dto instance from the request
*/
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeInfo.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeInfo.java
index 3772f67d7..007ad20f5 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeInfo.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeInfo.java
@@ -18,6 +18,7 @@
package org.apache.inlong.manager.pojo.node.redis;
import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
@@ -36,6 +37,42 @@ import org.apache.inlong.manager.pojo.node.DataNodeInfo;
@ApiModel("Redis data node info")
public class RedisDataNodeInfo extends DataNodeInfo {
+ /**
+ * Redis cluster mode
+ */
+ @ApiModelProperty(value = "Redis cluster mode")
+ private String clusterMode;
+
+ /**
+ * Redis host
+ */
+ @ApiModelProperty(value = "Redis host")
+ private String host;
+
+ /**
+ * Redis port
+ */
+ @ApiModelProperty(value = "Redis port")
+ private Integer port;
+
+ /**
+ * Redis sentinel master name
+ */
+ @ApiModelProperty(value = "Redis sentinel master name")
+ private String masterName;
+
+ /**
+ * Redis sentinel info
+ */
+ @ApiModelProperty(value = "Redis sentinel info")
+ private String sentinelsInfo;
+
+ /**
+ * Redis cluster nodes
+ */
+ @ApiModelProperty(value = "Redis cluster nodes")
+ private String clusterNodes;
+
public RedisDataNodeInfo() {
this.setType(DataNodeType.REDIS);
}
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeRequest.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeRequest.java
index 5d782c6f0..10028789e 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeRequest.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/redis/RedisDataNodeRequest.java
@@ -18,6 +18,7 @@
package org.apache.inlong.manager.pojo.node.redis;
import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
@@ -35,6 +36,41 @@ import org.apache.inlong.manager.pojo.node.DataNodeRequest;
@ApiModel("Redis data node request")
public class RedisDataNodeRequest extends DataNodeRequest {
+ /**
+ * Redis cluster mode
+ */
+ @ApiModelProperty(value = "Redis cluster mode")
+ private String clusterMode;
+
+ /**
+ * Redis host
+ */
+ @ApiModelProperty(value = "Redis host")
+ private String host;
+
+ /**
+ * Redis port
+ */
+ @ApiModelProperty(value = "Redis port")
+ private Integer port;
+
+ /**
+ * Redis sentinel master name
+ */
+ @ApiModelProperty(value = "Redis sentinel master name")
+ private String masterName;
+
+ /**
+ * Redis sentinel info
+ */
+ @ApiModelProperty(value = "Redis sentinel info")
+ private String sentinelsInfo;
+
+ /**
+ * Redis cluster nodes
+ */
+ @ApiModelProperty(value = "Redis cluster nodes")
+ private String clusterNodes;
public RedisDataNodeRequest() {
this.setType(DataNodeType.REDIS);
}
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSink.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSink.java
index ff19f870f..44b05bc08 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSink.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSink.java
@@ -88,7 +88,7 @@ public class RedisSink extends StreamSink {
private Integer port;
@ApiModelProperty("The master name of Redis sentinel cluster")
- private String sentinelMasterName;
+ private String masterName;
@ApiModelProperty("The sentinels info of Redis sentinel cluster")
private String sentinelsInfo;
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkDTO.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkDTO.java
index 22888179d..43bafde05 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkDTO.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkDTO.java
@@ -85,7 +85,7 @@ public class RedisSinkDTO {
private Integer port;
@ApiModelProperty("The master name of Redis sentinel cluster")
- private String sentinelMasterName;
+ private String masterName;
private String sentinelsInfo;
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkRequest.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkRequest.java
index 0ea2e84c9..76360c9ae 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkRequest.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/redis/RedisSinkRequest.java
@@ -82,7 +82,7 @@ public class RedisSinkRequest extends SinkRequest {
private Integer port;
@ApiModelProperty("The master name of Redis sentinel cluster")
- private String sentinelMasterName;
+ private String masterName;
@ApiModelProperty("The sentinels info of Redis sentinel cluster")
private String sentinelsInfo;
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/LoadNodeUtils.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/LoadNodeUtils.java
index fa32b79ab..0fbb69b16 100644
---
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/LoadNodeUtils.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/LoadNodeUtils.java
@@ -423,7 +423,7 @@ public class LoadNodeUtils {
String host = redisSink.getHost();
Integer port = redisSink.getPort();
String clusterNodes = redisSink.getClusterNodes();
- String sentinelMasterName = redisSink.getSentinelMasterName();
+ String masterName = redisSink.getMasterName();
String sentinelsInfo = redisSink.getSentinelsInfo();
Integer database = redisSink.getDatabase();
String password = redisSink.getPassword();
@@ -456,7 +456,7 @@ public class LoadNodeUtils {
host,
port,
clusterNodes,
- sentinelMasterName,
+ masterName,
sentinelsInfo,
database,
password,
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/redis/RedisDataNodeOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/redis/RedisDataNodeOperator.java
index 94930b4cd..818e33a4a 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/redis/RedisDataNodeOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/redis/RedisDataNodeOperator.java
@@ -23,21 +23,27 @@ import org.apache.inlong.manager.common.consts.DataNodeType;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.exceptions.BusinessException;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.DataNodeEntity;
import org.apache.inlong.manager.pojo.node.DataNodeInfo;
import org.apache.inlong.manager.pojo.node.DataNodeRequest;
import org.apache.inlong.manager.pojo.node.redis.RedisDataNodeDTO;
import org.apache.inlong.manager.pojo.node.redis.RedisDataNodeInfo;
import org.apache.inlong.manager.pojo.node.redis.RedisDataNodeRequest;
+import org.apache.inlong.manager.pojo.sink.redis.RedisClusterMode;
import org.apache.inlong.manager.service.node.AbstractDataNodeOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import static org.apache.inlong.manager.common.util.Preconditions.expectTrue;
+
@Service
public class RedisDataNodeOperator extends AbstractDataNodeOperator {
+ private static final int PORT_MAX_VALUE = 65535;
+
private static final Logger LOGGER =
LoggerFactory.getLogger(RedisDataNodeOperator.class);
@Autowired
@@ -71,6 +77,35 @@ public class RedisDataNodeOperator extends
AbstractDataNodeOperator {
@Override
protected void setTargetEntity(DataNodeRequest request, DataNodeEntity
targetEntity) {
RedisDataNodeRequest redisDataNodeRequest = (RedisDataNodeRequest)
request;
+
+ RedisClusterMode clusterMode =
RedisClusterMode.of(redisDataNodeRequest.getClusterMode());
+
+ switch (clusterMode) {
+ case STANDALONE:
+ String host = redisDataNodeRequest.getHost();
+ Preconditions.expectNotBlank(host, "Redis host cannot be
empty");
+ Integer port = redisDataNodeRequest.getPort();
+ expectTrue(
+ port != null && port > 1 && port < PORT_MAX_VALUE,
+ "The port of the redis server must be greater than 1
and less than " + PORT_MAX_VALUE +
+ "!");
+
+ break;
+ case SENTINEL:
+ String sentinelMasterName =
redisDataNodeRequest.getMasterName();
+ Preconditions.expectNotBlank(sentinelMasterName, "Redis
sentinel masterName cannot be empty");
+ String sentinelsInfo = redisDataNodeRequest.getSentinelsInfo();
+ Preconditions.expectNotBlank(sentinelsInfo, "Redis
sentinelsInfo cannot be empty");
+ break;
+ case CLUSTER:
+ String clusterNodes = redisDataNodeRequest.getClusterNodes();
+ Preconditions.expectNotBlank(clusterNodes, "Redis clusterNodes
cannot be empty");
+ break;
+
+ default:
+ throw new
BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT, "Unknown Redis cluster
mode");
+ }
+
CommonBeanUtils.copyProperties(redisDataNodeRequest, targetEntity,
true);
try {
RedisDataNodeDTO dto =
RedisDataNodeDTO.getFromRequest(redisDataNodeRequest);
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/redis/RedisSinkOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/redis/RedisSinkOperator.java
index 884fc64fd..3e5be4272 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/redis/RedisSinkOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/redis/RedisSinkOperator.java
@@ -94,7 +94,7 @@ public class RedisSinkOperator extends AbstractSinkOperator {
checkClusterNodes(clusterNodes);
break;
case SENTINEL:
- String sentinelMasterName =
sinkRequest.getSentinelMasterName();
+ String sentinelMasterName = sinkRequest.getMasterName();
expectNotEmpty(sentinelMasterName, "Redis MasterName of
Sentinel cluster must not null!");
String sentinelsInfo = sinkRequest.getSentinelsInfo();
expectNotEmpty(sentinelsInfo, "Redis sentinelsInfo of Sentinel
cluster must not null!");
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/RedisLoadNode.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/RedisLoadNode.java
index f17dd3602..4a7c1e200 100644
---
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/RedisLoadNode.java
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/RedisLoadNode.java
@@ -84,7 +84,7 @@ public class RedisLoadNode extends LoadNode implements
InlongMetric, Serializabl
private String host;
private Integer port;
private String clusterNodes;
- private String sentinelMasterName;
+ private String masterName;
private String sentinelsInfo;
private Integer database;
private String password;
@@ -117,7 +117,7 @@ public class RedisLoadNode extends LoadNode implements
InlongMetric, Serializabl
@Nullable @JsonProperty("host") String host,
@Nullable @JsonProperty("port") Integer port,
@Nullable @JsonProperty("clusterNodes") String clusterNodes,
- @Nullable @JsonProperty("sentinelMasterName") String
sentinelMasterName,
+ @Nullable @JsonProperty("masterName") String masterName,
@Nullable @JsonProperty("sentinelsInfo") String sentinelsInfo,
@Nullable @JsonProperty("database") Integer database,
@Nullable @JsonProperty("password") String password,
@@ -137,7 +137,7 @@ public class RedisLoadNode extends LoadNode implements
InlongMetric, Serializabl
this.host = host;
this.port = port;
this.clusterNodes = clusterNodes;
- this.sentinelMasterName = sentinelMasterName;
+ this.masterName = masterName;
this.sentinelsInfo = sentinelsInfo;
this.database = database;
this.password = password;
@@ -172,7 +172,7 @@ public class RedisLoadNode extends LoadNode implements
InlongMetric, Serializabl
options.put(CLUSTER_PASSWORD, password);
}
} else {
- options.put(MASTER_NAME, sentinelMasterName);
+ options.put(MASTER_NAME, masterName);
options.put(SENTINELS_INFO, sentinelsInfo);
if (StringUtils.isNotBlank(password)) {
options.put(SENTINELS_PASSWORD, password);
