[INLONG-11487][Manager] Support adding data add tasks based on the so…

…urce ID

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/DataAddTaskRequest.java

@@ -24,6 +24,7 @@
 import org.hibernate.validator.constraints.Length;
 
 import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
 
 import java.util.List;
 
@@ -39,7 +40,7 @@ public class DataAddTaskRequest {
     @NotBlank(message = "inlongGroupId cannot be blank")
     private String groupId;
 
-    @ApiModelProperty(value = "Source ID", hidden = true)
+    @ApiModelProperty(value = "Source ID, When the source ID is not null, data add task entries based on the source ID")
     private Integer sourceId;
 
     @ApiModelProperty(value = "Agent ip List")
@@ -53,4 +54,14 @@ public class DataAddTaskRequest {
     @ApiModelProperty(value = "Audit version", hidden = true)
     private String auditVersion;
 
+    @ApiModelProperty(value = "Data start time")
+    private String dataTimeFrom;
+
+    @ApiModelProperty(value = "Data end time")
+    private String dataTimeTo;
+
+    @ApiModelProperty(value = "Increase Audit Version")
+    @NotNull(message = "IncreaseAuditVersion cannot be null")
+    private Boolean increaseAuditVersion;
+
 }

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileDataAddTaskRequest.java

@@ -39,10 +39,4 @@ public class FileDataAddTaskRequest extends DataAddTaskRequest {
     @ApiModelProperty("filterStreams")
     private List<String> filterStreams;
 
-    @ApiModelProperty("Start time")
-    private Long startTime;
-
-    @ApiModelProperty("End time")
-    private Long endTime;
-
 }

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSource.java

@@ -89,11 +89,11 @@ public class FileSource extends StreamSource {
     @ApiModelProperty("Whether retry")
     private Boolean retry;
 
-    @ApiModelProperty("Start time")
-    private Long startTime;
+    @ApiModelProperty(value = "Data start time")
+    private String dataTimeFrom;
 
-    @ApiModelProperty("End time")
-    private Long endTime;
+    @ApiModelProperty(value = "Data end time")
+    private String dataTimeTo;
 
     @ApiModelProperty("filterStreams")
     private List<String> filterStreams;

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSourceDTO.java

@@ -92,11 +92,11 @@ public class FileSourceDTO {
     @ApiModelProperty("Whether retry")
     private Boolean retry = false;
 
-    @ApiModelProperty("Start time")
-    private Long startTime = 0L;
+    @ApiModelProperty(value = "Data start time")
+    private String dataTimeFrom;
 
-    @ApiModelProperty("End time")
-    private Long endTime = 0L;
+    @ApiModelProperty(value = "Data end time")
+    private String dataTimeTo;
 
     @ApiModelProperty(value = "Audit version")
     private String auditVersion;

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/file/FileSourceRequest.java

@@ -84,12 +84,11 @@ public class FileSourceRequest extends SourceRequest {
     @ApiModelProperty("Whether retry")
     private Boolean retry;
 
-    @ApiModelProperty("Start time")
-    private Long startTime;
-
-    @ApiModelProperty("End time")
-    private Long endTime;
+    @ApiModelProperty(value = "Data start time")
+    private String dataTimeFrom;
 
+    @ApiModelProperty(value = "Data end time")
+    private String dataTimeTo;
     @ApiModelProperty("filterStreams")
     private List<String> filterStreams;
 

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/node/AgentClusterNodeInstallOperator.java

@@ -59,6 +59,7 @@ public class AgentClusterNodeInstallOperator implements InlongClusterNodeInstall
 
     public static final String INSTALLER_CONF_PATH = "/conf/installer.properties";
     public static final String INSTALLER_START_CMD = "/bin/installer.sh start";
+    public static final String CRONTAB_START_CMD = "/bin/crontab.sh";
     public static final String INSTALLER_RESTART_CMD = "/bin/installer.sh restart";
     public static final String INSTALLER_STOP_CMD = "/bin/installer.sh restart";
     public static final String AGENT_MANAGER_AUTH_SECRET_ID = "agent.manager.auth.secretId";
@@ -115,6 +116,8 @@ public boolean install(ClusterNodeRequest clusterNodeRequest, String operator) {
             deployInstaller(request, operator);
             String startCmd = agentInstallPath + INSTALLER_START_CMD;
             commandExecutor.execRemote(request, startCmd);
+            String crontabStartCmd = agentInstallPath + CRONTAB_START_CMD;
+            commandExecutor.execRemote(request, crontabStartCmd);
             clusterNodeEntityMapper.updateOperateLogById(clusterNodeRequest.getId(),
                     NodeStatus.INSTALL_SUCCESS.getStatus(), currentTime + InlongConstants.BLANK + "success to install");
         } catch (Exception e) {
@@ -149,6 +152,8 @@ public boolean reInstall(ClusterNodeRequest clusterNodeRequest, String operator)
             commandExecutor.cpDir(request, agentInstallTempPath + "/modules.json", agentInstallPath + "/conf");
             String reStartCmd = agentInstallPath + INSTALLER_RESTART_CMD;
             commandExecutor.execRemote(request, reStartCmd);
+            String crontabStartCmd = agentInstallPath + CRONTAB_START_CMD;
+            commandExecutor.execRemote(request, crontabStartCmd);
             clusterNodeEntityMapper.updateOperateLogById(clusterNodeRequest.getId(), NodeStatus.NORMAL.getStatus(),
                     currentTime + InlongConstants.BLANK + "success to reinstall");
         } catch (Exception e) {

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutorImpl.java

@@ -147,7 +147,8 @@ public CommandResult rmDir(AgentClusterNodeRequest clusterNodeRequest, String pa
     @Override
     public CommandResult cpDir(AgentClusterNodeRequest clusterNodeRequest, String sourcePath, String targetPath)
             throws Exception {
-        return execRemote(clusterNodeRequest, "cp " + sourcePath + " " + targetPath);
+        return execRemote(clusterNodeRequest,
+                "if [ -e " + sourcePath + " ]; then cp " + sourcePath + " " + targetPath + "; fi");
     }
 
 }

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java

@@ -545,12 +545,21 @@ public List<Integer> addDataAddTask(DataAddTaskRequest request, String operator)
         List<String> agentIpList = request.getAgentIpList();
         List<StreamSourceEntity> entityList = new ArrayList<>();
         List<Integer> resultIdList = new ArrayList<>();
-        if (CollectionUtils.isEmpty(agentIpList)) {
-            entityList = sourceMapper.selectByRelatedId(request.getGroupId(), null, null);
+        if (request.getSourceId() != null) {
+            StreamSourceEntity entity = sourceMapper.selectById(request.getSourceId());
+            Preconditions.expectNotNull(entity, ErrorCodeEnum.SOURCE_INFO_NOT_FOUND);
+            entityList.add(entity);
         } else {
-            for (String agentIp : agentIpList) {
-                List<StreamSourceEntity> sourceEntityList = sourceMapper.selectByAgentIp(agentIp);
-                entityList.addAll(sourceEntityList);
+            if (agentIpList == null) {
+                throw new BusinessException("Agent ip list can not null");
+            }
+            if (CollectionUtils.isEmpty(agentIpList)) {
+                entityList = sourceMapper.selectByRelatedId(request.getGroupId(), null, null);
+            } else {
+                for (String agentIp : agentIpList) {
+                    List<StreamSourceEntity> sourceEntityList = sourceMapper.selectByAgentIp(agentIp);
+                    entityList.addAll(sourceEntityList);
+                }
             }
         }
         for (StreamSourceEntity sourceEntity : entityList) {

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/file/FileSourceOperator.java

@@ -114,10 +114,12 @@ public Integer addDataAddTask(DataAddTaskRequest request, String operator) {
         try {
             List<StreamSourceEntity> dataAddTaskList = sourceMapper.selectByTaskMapId(sourceEntity.getId());
             FileSourceDTO dto = FileSourceDTO.getFromJson(sourceEntity.getExtParams());
-            dto.setStartTime(sourceRequest.getStartTime());
-            dto.setEndTime(sourceRequest.getEndTime());
+            dto.setDataTimeFrom(sourceRequest.getDataTimeFrom());
+            dto.setDataTimeTo(sourceRequest.getDataTimeTo());
             dto.setRetry(true);
-            dto.setAuditVersion(request.getAuditVersion());
+            if (request.getIncreaseAuditVersion()) {
+                dto.setAuditVersion(request.getAuditVersion());
+            }
             dto.setFilterStreams(sourceRequest.getFilterStreams());
             StreamSourceEntity dataAddTaskEntity =
                     CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new);