| /* |
| * 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. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.inlong.manager.service.core.impl; |
| |
| import com.google.common.base.Joiner; |
| import com.google.common.collect.Lists; |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import com.google.gson.Gson; |
| import org.apache.commons.collections.CollectionUtils; |
| import org.apache.commons.lang3.RandomStringUtils; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.inlong.common.constant.Constants; |
| import org.apache.inlong.common.constant.MQType; |
| import org.apache.inlong.common.db.CommandEntity; |
| import org.apache.inlong.common.enums.PullJobTypeEnum; |
| import org.apache.inlong.common.enums.TaskTypeEnum; |
| import org.apache.inlong.common.pojo.agent.CmdConfig; |
| import org.apache.inlong.common.pojo.agent.DataConfig; |
| import org.apache.inlong.common.pojo.agent.TaskRequest; |
| import org.apache.inlong.common.pojo.agent.TaskResult; |
| import org.apache.inlong.common.pojo.agent.TaskSnapshotRequest; |
| import org.apache.inlong.common.pojo.dataproxy.DataProxyTopicInfo; |
| import org.apache.inlong.common.pojo.dataproxy.MQClusterInfo; |
| import org.apache.inlong.manager.common.consts.InlongConstants; |
| import org.apache.inlong.manager.common.consts.SourceType; |
| import org.apache.inlong.manager.common.enums.ClusterType; |
| import org.apache.inlong.manager.common.enums.GroupStatus; |
| import org.apache.inlong.manager.common.enums.SourceStatus; |
| import org.apache.inlong.manager.common.exceptions.BusinessException; |
| import org.apache.inlong.manager.common.util.CommonBeanUtils; |
| import org.apache.inlong.manager.common.util.JsonUtils; |
| import org.apache.inlong.manager.common.util.Preconditions; |
| import org.apache.inlong.manager.dao.entity.InlongClusterEntity; |
| import org.apache.inlong.manager.dao.entity.InlongClusterNodeEntity; |
| import org.apache.inlong.manager.dao.entity.InlongGroupEntity; |
| import org.apache.inlong.manager.dao.entity.InlongStreamEntity; |
| import org.apache.inlong.manager.dao.entity.StreamSourceEntity; |
| import org.apache.inlong.manager.dao.mapper.DataSourceCmdConfigEntityMapper; |
| import org.apache.inlong.manager.dao.mapper.InlongClusterEntityMapper; |
| import org.apache.inlong.manager.dao.mapper.InlongClusterNodeEntityMapper; |
| import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper; |
| import org.apache.inlong.manager.dao.mapper.InlongStreamEntityMapper; |
| import org.apache.inlong.manager.dao.mapper.StreamSourceEntityMapper; |
| import org.apache.inlong.manager.pojo.cluster.ClusterPageRequest; |
| import org.apache.inlong.manager.pojo.cluster.agent.AgentClusterNodeBindGroupRequest; |
| import org.apache.inlong.manager.pojo.cluster.agent.AgentClusterNodeDTO; |
| import org.apache.inlong.manager.pojo.cluster.pulsar.PulsarClusterDTO; |
| import org.apache.inlong.manager.pojo.group.pulsar.InlongPulsarDTO; |
| import org.apache.inlong.manager.pojo.source.file.FileSourceDTO; |
| import org.apache.inlong.manager.service.core.AgentService; |
| import org.apache.inlong.manager.service.source.SourceSnapshotOperator; |
| import org.elasticsearch.common.util.set.Sets; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.springframework.beans.factory.annotation.Autowired; |
| import org.springframework.beans.factory.annotation.Value; |
| import org.springframework.stereotype.Service; |
| import org.springframework.transaction.annotation.Isolation; |
| import org.springframework.transaction.annotation.Propagation; |
| import org.springframework.transaction.annotation.Transactional; |
| |
| import javax.annotation.PostConstruct; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.concurrent.ArrayBlockingQueue; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ThreadFactory; |
| import java.util.concurrent.ThreadPoolExecutor; |
| import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.Collectors; |
| import java.util.stream.Stream; |
| |
| import static org.apache.inlong.manager.common.consts.InlongConstants.DOT; |
| |
| /** |
| * Agent service layer implementation |
| */ |
| @Service |
| public class AgentServiceImpl implements AgentService { |
| |
| private static final Logger LOGGER = LoggerFactory.getLogger(AgentServiceImpl.class); |
| private static final int UNISSUED_STATUS = 2; |
| private static final int ISSUED_STATUS = 3; |
| private static final int MODULUS_100 = 100; |
| private static final int TASK_FETCH_SIZE = 2; |
| private static final Gson GSON = new Gson(); |
| private final ExecutorService executorService = new ThreadPoolExecutor( |
| 5, |
| 10, |
| 10L, |
| TimeUnit.SECONDS, |
| new ArrayBlockingQueue<>(100), |
| new ThreadFactoryBuilder().setNameFormat("async-agent-%s").build(), |
| new CallerRunsPolicy()); |
| @Value("${source.update.enabled:false}") |
| private Boolean updateTaskTimeoutEnabled; |
| @Value("${source.update.before.seconds:60}") |
| private Integer beforeSeconds; |
| @Value("${source.update.interval:60}") |
| private Integer updateTaskInterval; |
| @Value("${source.cleansing.enabled:false}") |
| private Boolean sourceCleanEnabled; |
| @Value("${source.cleansing.interval:600}") |
| private Integer cleanInterval; |
| @Autowired |
| private StreamSourceEntityMapper sourceMapper; |
| @Autowired |
| private SourceSnapshotOperator snapshotOperator; |
| @Autowired |
| private DataSourceCmdConfigEntityMapper sourceCmdConfigMapper; |
| @Autowired |
| private InlongGroupEntityMapper groupMapper; |
| @Autowired |
| private InlongStreamEntityMapper streamMapper; |
| @Autowired |
| private InlongClusterEntityMapper clusterMapper; |
| @Autowired |
| private InlongClusterNodeEntityMapper clusterNodeMapper; |
| |
| /** |
| * Start the update task |
| */ |
| @PostConstruct |
| private void startHeartbeatTask() { |
| if (updateTaskTimeoutEnabled) { |
| ThreadFactory factory = new ThreadFactoryBuilder() |
| .setNameFormat("scheduled-source-timeout-%d") |
| .setDaemon(true) |
| .build(); |
| ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(factory); |
| executor.scheduleWithFixedDelay(() -> { |
| try { |
| sourceMapper.updateStatusToTimeout(beforeSeconds); |
| LOGGER.info("update task status successfully"); |
| } catch (Throwable t) { |
| LOGGER.error("update task status error", t); |
| } |
| }, 0, updateTaskInterval, TimeUnit.SECONDS); |
| LOGGER.info("update task status started successfully"); |
| } |
| if (sourceCleanEnabled) { |
| ThreadFactory factory = new ThreadFactoryBuilder() |
| .setNameFormat("scheduled-source-deleted-%d") |
| .setDaemon(true) |
| .build(); |
| ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(factory); |
| executor.scheduleWithFixedDelay(() -> { |
| try { |
| sourceMapper.updateStatusByDeleted(); |
| LOGGER.info("clean task successfully"); |
| } catch (Throwable t) { |
| LOGGER.error("clean task error", t); |
| } |
| }, 0, cleanInterval, TimeUnit.SECONDS); |
| LOGGER.info("clean task started successfully"); |
| } |
| } |
| |
| @Override |
| public Boolean reportSnapshot(TaskSnapshotRequest request) { |
| return snapshotOperator.snapshot(request); |
| } |
| |
| @Override |
| @Transactional(rollbackFor = Throwable.class, isolation = Isolation.READ_COMMITTED, propagation = Propagation.REQUIRES_NEW) |
| public void report(TaskRequest request) { |
| if (LOGGER.isDebugEnabled()) { |
| LOGGER.debug("begin to get agent task: {}", request); |
| } |
| if (request == null || StringUtils.isBlank(request.getAgentIp())) { |
| throw new BusinessException("agent request or agent ip was empty, just return"); |
| } |
| |
| preTimeoutTasks(request); |
| |
| // Update task status, other tasks with status 20x will change to 30x in next request |
| if (CollectionUtils.isEmpty(request.getCommandInfo())) { |
| LOGGER.info("task result was empty in request: {}, just return", request); |
| return; |
| } |
| for (CommandEntity command : request.getCommandInfo()) { |
| updateTaskStatus(command); |
| } |
| } |
| |
| /** |
| * Update task status by command. |
| * |
| * @param command command info. |
| */ |
| private void updateTaskStatus(CommandEntity command) { |
| Integer taskId = command.getTaskId(); |
| StreamSourceEntity current = sourceMapper.selectForAgentTask(taskId); |
| if (current == null) { |
| LOGGER.warn("stream source not found by id={}, just return", taskId); |
| return; |
| } |
| |
| if (!Objects.equals(command.getVersion(), current.getVersion())) { |
| LOGGER.warn("task result version [{}] not equals to current [{}] for id [{}], skip update", |
| command.getVersion(), current.getVersion(), taskId); |
| return; |
| } |
| |
| int result = command.getCommandResult(); |
| int previousStatus = current.getStatus(); |
| int nextStatus = SourceStatus.SOURCE_NORMAL.getCode(); |
| |
| if (Constants.RESULT_FAIL == result) { |
| LOGGER.warn("task failed for id =[{}]", taskId); |
| nextStatus = SourceStatus.SOURCE_FAILED.getCode(); |
| } else if (previousStatus / MODULUS_100 == ISSUED_STATUS) { |
| // Change the status from 30x to normal / disable / frozen |
| if (SourceStatus.BEEN_ISSUED_DELETE.getCode() == previousStatus) { |
| nextStatus = SourceStatus.SOURCE_DISABLE.getCode(); |
| } else if (SourceStatus.BEEN_ISSUED_STOP.getCode() == previousStatus) { |
| nextStatus = SourceStatus.SOURCE_STOP.getCode(); |
| } |
| } |
| |
| if (nextStatus != previousStatus) { |
| sourceMapper.updateStatus(taskId, nextStatus, false); |
| LOGGER.info("task result=[{}], update source status to [{}] for id [{}]", result, nextStatus, taskId); |
| } |
| } |
| |
| @Override |
| @Transactional(rollbackFor = Throwable.class, isolation = Isolation.READ_COMMITTED, propagation = Propagation.REQUIRES_NEW) |
| public TaskResult getTaskResult(TaskRequest request) { |
| if (StringUtils.isBlank(request.getClusterName()) || StringUtils.isBlank(request.getAgentIp())) { |
| throw new BusinessException("agent request or agent ip was empty, just return"); |
| } |
| |
| preProcessFileTask(request); |
| preProcessNonFileTasks(request); |
| List<DataConfig> tasks = processQueuedTasks(request); |
| |
| // Query pending special commands |
| List<CmdConfig> cmdConfigs = getAgentCmdConfigs(request); |
| return TaskResult.builder().dataConfigs(tasks).cmdConfigs(cmdConfigs).build(); |
| } |
| |
| @Override |
| @Transactional(rollbackFor = Throwable.class, isolation = Isolation.READ_COMMITTED, propagation = Propagation.REQUIRES_NEW) |
| public Boolean bindGroup(AgentClusterNodeBindGroupRequest request) { |
| HashSet<String> bindSet = Sets.newHashSet(); |
| HashSet<String> unbindSet = Sets.newHashSet(); |
| if (request.getBindClusterNodes() != null) { |
| bindSet.addAll(request.getBindClusterNodes()); |
| } |
| if (request.getUnbindClusterNodes() != null) { |
| unbindSet.addAll(request.getUnbindClusterNodes()); |
| } |
| Preconditions.expectTrue(Sets.union(bindSet, unbindSet).size() == bindSet.size() + unbindSet.size(), |
| "can not add and del node tag in the sameTime"); |
| InlongClusterEntity cluster = clusterMapper.selectByNameAndType(request.getClusterName(), ClusterType.AGENT); |
| String message = "Current user does not have permission to bind cluster node tag"; |
| |
| if (CollectionUtils.isNotEmpty(bindSet)) { |
| bindSet.stream().flatMap(clusterNode -> { |
| ClusterPageRequest pageRequest = new ClusterPageRequest(); |
| pageRequest.setParentId(cluster.getId()); |
| pageRequest.setType(ClusterType.AGENT); |
| pageRequest.setKeyword(clusterNode); |
| return clusterNodeMapper.selectByCondition(pageRequest).stream(); |
| }).filter(Objects::nonNull) |
| .forEach(entity -> { |
| Set<String> groupSet = new HashSet<>(); |
| AgentClusterNodeDTO agentClusterNodeDTO = new AgentClusterNodeDTO(); |
| if (StringUtils.isNotBlank(entity.getExtParams())) { |
| agentClusterNodeDTO = AgentClusterNodeDTO.getFromJson(entity.getExtParams()); |
| String agentGroup = agentClusterNodeDTO.getAgentGroup(); |
| groupSet = StringUtils.isBlank(agentGroup) ? groupSet |
| : Sets.newHashSet(agentGroup.split(InlongConstants.COMMA)); |
| } |
| groupSet.add(request.getAgentGroup()); |
| agentClusterNodeDTO.setAgentGroup(Joiner.on(",").join(groupSet)); |
| entity.setExtParams(GSON.toJson(agentClusterNodeDTO)); |
| clusterNodeMapper.insertOnDuplicateKeyUpdate(entity); |
| }); |
| } |
| |
| if (CollectionUtils.isNotEmpty(unbindSet)) { |
| unbindSet.stream().flatMap(clusterNode -> { |
| ClusterPageRequest pageRequest = new ClusterPageRequest(); |
| pageRequest.setParentId(cluster.getId()); |
| pageRequest.setType(ClusterType.AGENT); |
| pageRequest.setKeyword(clusterNode); |
| return clusterNodeMapper.selectByCondition(pageRequest).stream(); |
| }).filter(Objects::nonNull) |
| .forEach(entity -> { |
| Set<String> groupSet = new HashSet<>(); |
| AgentClusterNodeDTO agentClusterNodeDTO = new AgentClusterNodeDTO(); |
| if (StringUtils.isNotBlank(entity.getExtParams())) { |
| agentClusterNodeDTO = AgentClusterNodeDTO.getFromJson(entity.getExtParams()); |
| String agentGroup = agentClusterNodeDTO.getAgentGroup(); |
| groupSet = StringUtils.isBlank(agentGroup) ? groupSet |
| : Sets.newHashSet(agentGroup.split(InlongConstants.COMMA)); |
| } |
| groupSet.remove(request.getAgentGroup()); |
| agentClusterNodeDTO.setAgentGroup(Joiner.on(",").join(groupSet)); |
| entity.setExtParams(GSON.toJson(agentClusterNodeDTO)); |
| clusterNodeMapper.insertOnDuplicateKeyUpdate(entity); |
| }); |
| } |
| return true; |
| } |
| |
| /** |
| * Query the tasks that source is waited to be operated.(only clusterName and ip matched it can be operated) |
| */ |
| private List<DataConfig> processQueuedTasks(TaskRequest request) { |
| HashSet<SourceStatus> needAddStatusSet = Sets.newHashSet(SourceStatus.TOBE_ISSUED_SET); |
| if (PullJobTypeEnum.NEVER == PullJobTypeEnum.getPullJobType(request.getPullJobType())) { |
| LOGGER.warn("agent pull job type is [NEVER], just pull to be active tasks"); |
| needAddStatusSet.remove(SourceStatus.TO_BE_ISSUED_ADD); |
| } |
| |
| // todo:Find out why uuid is necessary here, if it is invalid, ignore it |
| List<StreamSourceEntity> sourceEntities = sourceMapper.selectByStatusAndCluster( |
| needAddStatusSet.stream().map(SourceStatus::getCode).collect(Collectors.toList()), |
| request.getClusterName(), request.getAgentIp(), request.getUuid()); |
| List<DataConfig> issuedTasks = Lists.newArrayList(); |
| for (StreamSourceEntity sourceEntity : sourceEntities) { |
| int op = getOp(sourceEntity.getStatus()); |
| int nextStatus = getNextStatus(sourceEntity.getStatus()); |
| sourceEntity.setPreviousStatus(sourceEntity.getStatus()); |
| sourceEntity.setStatus(nextStatus); |
| if (sourceMapper.updateByPrimaryKeySelective(sourceEntity) == 1) { |
| sourceEntity.setVersion(sourceEntity.getVersion() + 1); |
| DataConfig dataConfig = getDataConfig(sourceEntity, op); |
| issuedTasks.add(dataConfig); |
| LOGGER.info("Offer source task({}) for agent({}) in cluster({})", |
| dataConfig, request.getAgentIp(), request.getClusterName()); |
| } |
| } |
| return issuedTasks; |
| } |
| |
| // todo:If many agents pull the same non-file task in this place, wouldn’t it be a problem? |
| // it will issue multiple tasks |
| private void preProcessNonFileTasks(TaskRequest taskRequest) { |
| List<Integer> needAddStatusList; |
| if (PullJobTypeEnum.NEVER == PullJobTypeEnum.getPullJobType(taskRequest.getPullJobType())) { |
| LOGGER.warn("agent pull job type is [NEVER], just pull to be active tasks"); |
| needAddStatusList = Collections.singletonList(SourceStatus.TO_BE_ISSUED_ACTIVE.getCode()); |
| } else { |
| needAddStatusList = Arrays.asList(SourceStatus.TO_BE_ISSUED_ADD.getCode(), |
| SourceStatus.TO_BE_ISSUED_ACTIVE.getCode()); |
| } |
| List<String> sourceTypes = Lists.newArrayList(SourceType.MYSQL_SQL, SourceType.KAFKA, |
| SourceType.MYSQL_BINLOG, SourceType.POSTGRESQL); |
| List<StreamSourceEntity> sourceEntities = sourceMapper.selectByStatusAndType(needAddStatusList, sourceTypes, |
| TASK_FETCH_SIZE); |
| for (StreamSourceEntity sourceEntity : sourceEntities) { |
| // refresh agent ip and uuid to make it can be processed in queued task |
| sourceEntity.setAgentIp(taskRequest.getAgentIp()); |
| sourceEntity.setUuid(taskRequest.getUuid()); |
| sourceMapper.updateByPrimaryKeySelective(sourceEntity); |
| } |
| } |
| |
| private void preProcessFileTask(TaskRequest taskRequest) { |
| preProcessTemplateFileTask(taskRequest); |
| preProcessLabelFileTasks(taskRequest); |
| } |
| |
| /** |
| * Add subtasks to template tasks. |
| * (Template task are agent_ip is null and template_id is null) |
| */ |
| private void preProcessTemplateFileTask(TaskRequest taskRequest) { |
| List<Integer> needCopiedStatusList = Arrays.asList(SourceStatus.TO_BE_ISSUED_ADD.getCode(), |
| SourceStatus.TO_BE_ISSUED_ACTIVE.getCode()); |
| final String agentIp = taskRequest.getAgentIp(); |
| final String agentClusterName = taskRequest.getClusterName(); |
| Preconditions.expectTrue(StringUtils.isNotBlank(agentIp) || StringUtils.isNotBlank(agentClusterName), |
| "both agent ip and cluster name are blank when fetching file task"); |
| |
| // find those node whose tag match stream_source tag and agent ip match stream_source agent ip |
| List<StreamSourceEntity> sourceEntities = sourceMapper.selectTemplateSourceByCluster(needCopiedStatusList, |
| Lists.newArrayList(SourceType.FILE), agentClusterName); |
| Set<GroupStatus> noNeedAddTask = Sets.newHashSet(GroupStatus.SUSPENDED, GroupStatus.SUSPENDING); |
| sourceEntities.forEach(sourceEntity -> { |
| InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId()); |
| if (groupEntity != null && noNeedAddTask.contains(GroupStatus.forCode(groupEntity.getStatus()))) { |
| return; |
| } |
| StreamSourceEntity subSource = sourceMapper.selectOneByTemplatedIdAndAgentIp(sourceEntity.getId(), |
| agentIp); |
| if (subSource == null) { |
| // if not, clone a subtask for this Agent. |
| // note: a new source name with random suffix is generated to adhere to the unique constraint |
| StreamSourceEntity fileEntity = |
| CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new); |
| fileEntity.setSourceName(fileEntity.getSourceName() + "-" |
| + RandomStringUtils.randomAlphanumeric(10).toLowerCase(Locale.ROOT)); |
| fileEntity.setTemplateId(sourceEntity.getId()); |
| fileEntity.setAgentIp(agentIp); |
| fileEntity.setStatus(SourceStatus.TO_BE_ISSUED_ADD.getCode()); |
| // create new sub source task |
| sourceMapper.insert(fileEntity); |
| LOGGER.info("Transform new template task({}) for agent({}) in cluster({}).", |
| fileEntity.getId(), taskRequest.getAgentIp(), taskRequest.getClusterName()); |
| } |
| }); |
| } |
| |
| /** |
| * Find file collecting task match those condition: |
| * 1.agent ip match |
| * 2.cluster name match |
| * Send the corresponding task action request according to the matching state of the tag and the current state |
| */ |
| private void preProcessLabelFileTasks(TaskRequest taskRequest) { |
| List<Integer> needProcessedStatusList = Arrays.asList( |
| SourceStatus.SOURCE_NORMAL.getCode(), |
| SourceStatus.SOURCE_FAILED.getCode(), |
| SourceStatus.SOURCE_STOP.getCode(), |
| SourceStatus.TO_BE_ISSUED_ADD.getCode(), |
| SourceStatus.TO_BE_ISSUED_STOP.getCode(), |
| SourceStatus.TO_BE_ISSUED_ACTIVE.getCode()); |
| final String agentIp = taskRequest.getAgentIp(); |
| final String agentClusterName = taskRequest.getClusterName(); |
| Preconditions.expectTrue(StringUtils.isNotBlank(agentIp) || StringUtils.isNotBlank(agentClusterName), |
| "both agent ip and cluster name are blank when fetching file task"); |
| |
| InlongClusterNodeEntity clusterNodeEntity = selectByIpAndCluster(agentClusterName, agentIp); |
| List<StreamSourceEntity> sourceEntities = sourceMapper.selectByAgentIpAndCluster(needProcessedStatusList, |
| Lists.newArrayList(SourceType.FILE), agentIp, agentClusterName); |
| |
| sourceEntities.forEach(sourceEntity -> { |
| // case: agent tag unbind and mismatch source task |
| Set<SourceStatus> exceptedUnmatchedStatus = Sets.newHashSet( |
| SourceStatus.SOURCE_STOP, |
| SourceStatus.TO_BE_ISSUED_STOP); |
| if (!matchGroup(sourceEntity, clusterNodeEntity) |
| && !exceptedUnmatchedStatus.contains(SourceStatus.forCode(sourceEntity.getStatus()))) { |
| LOGGER.info("Transform task({}) from {} to {} because tag mismatch " |
| + "for agent({}) in cluster({})", sourceEntity.getAgentIp(), |
| sourceEntity.getStatus(), SourceStatus.TO_BE_ISSUED_STOP.getCode(), |
| agentIp, agentClusterName); |
| sourceMapper.updateStatus( |
| sourceEntity.getId(), SourceStatus.TO_BE_ISSUED_STOP.getCode(), false); |
| } |
| |
| // case: agent tag rebind and match source task again and stream is not in 'SUSPENDED' status |
| InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId()); |
| Set<SourceStatus> exceptedMatchedSourceStatus = Sets.newHashSet( |
| SourceStatus.SOURCE_NORMAL, |
| SourceStatus.TO_BE_ISSUED_ADD, |
| SourceStatus.TO_BE_ISSUED_ACTIVE); |
| Set<GroupStatus> exceptedMatchedGroupStatus = Sets.newHashSet(GroupStatus.SUSPENDED, |
| GroupStatus.SUSPENDING); |
| if (matchGroup(sourceEntity, clusterNodeEntity) |
| && groupEntity != null |
| && !exceptedMatchedSourceStatus.contains(SourceStatus.forCode(sourceEntity.getStatus())) |
| && !exceptedMatchedGroupStatus.contains(GroupStatus.forCode(groupEntity.getStatus()))) { |
| LOGGER.info("Transform task({}) from {} to {} because tag rematch " |
| + "for agent({}) in cluster({})", sourceEntity.getAgentIp(), |
| sourceEntity.getStatus(), SourceStatus.TO_BE_ISSUED_ACTIVE.getCode(), |
| agentIp, agentClusterName); |
| sourceMapper.updateStatus( |
| sourceEntity.getId(), SourceStatus.TO_BE_ISSUED_ACTIVE.getCode(), false); |
| } |
| }); |
| } |
| |
| private void preTimeoutTasks(TaskRequest taskRequest) { |
| // If the agent report succeeds, restore the source status |
| List<Integer> needUpdateIds = sourceMapper.selectHeartbeatTimeoutIds(null, taskRequest.getAgentIp(), |
| taskRequest.getClusterName()); |
| // restore state for all source by ip and type |
| if (CollectionUtils.isNotEmpty(needUpdateIds)) { |
| sourceMapper.rollbackTimeoutStatusByIds(needUpdateIds, null); |
| } |
| } |
| |
| private InlongClusterNodeEntity selectByIpAndCluster(String clusterName, String ip) { |
| InlongClusterEntity clusterEntity = clusterMapper.selectByNameAndType(clusterName, ClusterType.AGENT); |
| if (clusterEntity == null) { |
| return null; |
| } |
| |
| ClusterPageRequest nodeRequest = new ClusterPageRequest(); |
| nodeRequest.setKeyword(ip); |
| nodeRequest.setParentId(clusterEntity.getId()); |
| nodeRequest.setType(ClusterType.AGENT); |
| return clusterNodeMapper.selectByCondition(nodeRequest).stream().findFirst().orElse(null); |
| } |
| |
| private int getOp(int status) { |
| return status % MODULUS_100; |
| } |
| |
| private int getNextStatus(int status) { |
| int op = status % MODULUS_100; |
| return ISSUED_STATUS * MODULUS_100 + op; |
| } |
| |
| /** |
| * Get the DataConfig from the stream source entity. |
| * |
| * @param entity stream source entity. |
| * @param op operation code for add, delete, etc. |
| * @return data config. |
| */ |
| private DataConfig getDataConfig(StreamSourceEntity entity, int op) { |
| DataConfig dataConfig = new DataConfig(); |
| dataConfig.setIp(entity.getAgentIp()); |
| dataConfig.setUuid(entity.getUuid()); |
| dataConfig.setOp(String.valueOf(op)); |
| dataConfig.setTaskId(entity.getId()); |
| dataConfig.setTaskType(getTaskType(entity)); |
| dataConfig.setTaskName(entity.getSourceName()); |
| dataConfig.setSnapshot(entity.getSnapshot()); |
| dataConfig.setVersion(entity.getVersion()); |
| |
| String groupId = entity.getInlongGroupId(); |
| String streamId = entity.getInlongStreamId(); |
| dataConfig.setInlongGroupId(groupId); |
| dataConfig.setInlongStreamId(streamId); |
| |
| InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId); |
| InlongStreamEntity streamEntity = streamMapper.selectByIdentifier(groupId, streamId); |
| String extParams = entity.getExtParams(); |
| if (groupEntity != null && streamEntity != null) { |
| dataConfig.setSyncSend(streamEntity.getSyncSend()); |
| if (SourceType.FILE.equalsIgnoreCase(streamEntity.getDataType())) { |
| String dataSeparator = streamEntity.getDataSeparator(); |
| extParams = (null != dataSeparator ? getExtParams(extParams, dataSeparator) : extParams); |
| } |
| |
| int dataReportType = groupEntity.getDataReportType(); |
| dataConfig.setDataReportType(dataReportType); |
| if (InlongConstants.REPORT_TO_MQ_RECEIVED == dataReportType) { |
| // add mq cluster setting |
| List<MQClusterInfo> mqSet = new ArrayList<>(); |
| List<String> clusterTagList = Collections.singletonList(groupEntity.getInlongClusterTag()); |
| ClusterPageRequest pageRequest = ClusterPageRequest.builder() |
| .type(groupEntity.getMqType()) |
| .clusterTagList(clusterTagList) |
| .build(); |
| List<InlongClusterEntity> mqClusterList = clusterMapper.selectByCondition(pageRequest); |
| for (InlongClusterEntity cluster : mqClusterList) { |
| MQClusterInfo clusterInfo = new MQClusterInfo(); |
| clusterInfo.setUrl(cluster.getUrl()); |
| clusterInfo.setToken(cluster.getToken()); |
| clusterInfo.setMqType(cluster.getType()); |
| clusterInfo.setParams(JsonUtils.parseObject(cluster.getExtParams(), HashMap.class)); |
| mqSet.add(clusterInfo); |
| } |
| dataConfig.setMqClusters(mqSet); |
| |
| // add topic setting |
| String mqResource = groupEntity.getMqResource(); |
| String mqType = groupEntity.getMqType(); |
| if (MQType.PULSAR.equals(mqType) || MQType.TDMQ_PULSAR.equals(mqType)) { |
| // first get the tenant from the InlongGroup, and then get it from the PulsarCluster. |
| InlongPulsarDTO pulsarDTO = InlongPulsarDTO.getFromJson(groupEntity.getExtParams()); |
| String tenant = pulsarDTO.getTenant(); |
| if (StringUtils.isBlank(tenant)) { |
| // If there are multiple Pulsar clusters, take the first one. |
| // Note that the tenants in multiple Pulsar clusters must be identical. |
| PulsarClusterDTO pulsarCluster = PulsarClusterDTO.getFromJson( |
| mqClusterList.get(0).getExtParams()); |
| tenant = pulsarCluster.getTenant(); |
| } |
| |
| String topic = String.format(InlongConstants.PULSAR_TOPIC_FORMAT, |
| tenant, mqResource, streamEntity.getMqResource()); |
| DataProxyTopicInfo topicConfig = new DataProxyTopicInfo(); |
| topicConfig.setInlongGroupId(groupId + "/" + streamId); |
| topicConfig.setTopic(topic); |
| dataConfig.setTopicInfo(topicConfig); |
| } else if (MQType.TUBEMQ.equals(mqType)) { |
| DataProxyTopicInfo topicConfig = new DataProxyTopicInfo(); |
| topicConfig.setInlongGroupId(groupId); |
| topicConfig.setTopic(mqResource); |
| dataConfig.setTopicInfo(topicConfig); |
| } else if (MQType.KAFKA.equals(mqType)) { |
| DataProxyTopicInfo topicConfig = new DataProxyTopicInfo(); |
| topicConfig.setInlongGroupId(groupId); |
| topicConfig.setTopic(groupEntity.getMqResource() + DOT + streamEntity.getMqResource()); |
| dataConfig.setTopicInfo(topicConfig); |
| } |
| } else { |
| LOGGER.warn("set syncSend=[0] as the stream not exists for groupId={}, streamId={}", groupId, streamId); |
| } |
| } |
| dataConfig.setExtParams(extParams); |
| return dataConfig; |
| } |
| |
| private String getExtParams(String extParams, String dataSeparator) { |
| FileSourceDTO fileSourceDTO = JsonUtils.parseObject(extParams, FileSourceDTO.class); |
| if (Objects.nonNull(fileSourceDTO)) { |
| fileSourceDTO.setDataSeparator(dataSeparator); |
| return JsonUtils.toJsonString(fileSourceDTO); |
| } |
| return extParams; |
| } |
| |
| /** |
| * Get the Task type from the stream source entity. |
| * |
| * @param sourceEntity stream source info. |
| * @return task type |
| */ |
| private int getTaskType(StreamSourceEntity sourceEntity) { |
| TaskTypeEnum taskType = SourceType.SOURCE_TASK_MAP.get(sourceEntity.getSourceType()); |
| if (taskType == null) { |
| throw new BusinessException("Unsupported task type for source type " + sourceEntity.getSourceType()); |
| } |
| return taskType.getType(); |
| } |
| |
| // todo:delete it, source cmd is useless |
| |
| /** |
| * Get the agent command config by the agent ip. |
| * |
| * @param taskRequest task request info. |
| * @return agent command config list. |
| */ |
| private List<CmdConfig> getAgentCmdConfigs(TaskRequest taskRequest) { |
| return sourceCmdConfigMapper.queryCmdByAgentIp(taskRequest.getAgentIp()).stream().map(cmd -> { |
| CmdConfig cmdConfig = new CmdConfig(); |
| cmdConfig.setDataTime(cmd.getSpecifiedDataTime()); |
| cmdConfig.setOp(cmd.getCmdType()); |
| cmdConfig.setId(cmd.getId()); |
| cmdConfig.setTaskId(cmd.getTaskId()); |
| return cmdConfig; |
| }).collect(Collectors.toList()); |
| } |
| |
| private boolean matchGroup(StreamSourceEntity sourceEntity, InlongClusterNodeEntity clusterNodeEntity) { |
| Preconditions.expectNotNull(sourceEntity, "cluster must be valid"); |
| if (sourceEntity.getInlongClusterNodeGroup() == null) { |
| return true; |
| } |
| |
| if (clusterNodeEntity == null || clusterNodeEntity.getExtParams() == null) { |
| return false; |
| } |
| |
| Set<String> clusterNodeGroups = new HashSet<>(); |
| if (StringUtils.isNotBlank(clusterNodeEntity.getExtParams())) { |
| AgentClusterNodeDTO agentClusterNodeDTO = AgentClusterNodeDTO.getFromJson(clusterNodeEntity.getExtParams()); |
| String agentGroup = agentClusterNodeDTO.getAgentGroup(); |
| clusterNodeGroups = StringUtils.isBlank(agentGroup) ? new HashSet<>() |
| : Sets.newHashSet(agentGroup.split(InlongConstants.COMMA)); |
| } |
| Set<String> sourceGroups = Stream.of( |
| sourceEntity.getInlongClusterNodeGroup().split(InlongConstants.COMMA)).collect(Collectors.toSet()); |
| return sourceGroups.stream().anyMatch(clusterNodeGroups::contains); |
| } |
| |
| } |
