Wenjun Ruan
7 months ago
committed by
GitHub
29 changed files with 484 additions and 122 deletions
@ -0,0 +1,39 @@
|
||||
/* |
||||
* 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.dolphinscheduler.dao.repository; |
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command; |
||||
|
||||
import java.util.List; |
||||
|
||||
public interface CommandDao extends IDao<Command> { |
||||
|
||||
/** |
||||
* Query command by command id and server slot, return the command which match (commandId / step) %s totalSlot = currentSlotIndex |
||||
* |
||||
* @param currentSlotIndex current slot index |
||||
* @param totalSlot total slot number |
||||
* @param idStep id step in db |
||||
* @param fetchNum fetch number |
||||
* @return command list |
||||
*/ |
||||
List<Command> queryCommandByIdSlot(int currentSlotIndex, |
||||
int totalSlot, |
||||
int idStep, |
||||
int fetchNum); |
||||
} |
||||
@ -0,0 +1,41 @@
|
||||
/* |
||||
* 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.dolphinscheduler.dao.repository.impl; |
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command; |
||||
import org.apache.dolphinscheduler.dao.mapper.CommandMapper; |
||||
import org.apache.dolphinscheduler.dao.repository.BaseDao; |
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao; |
||||
|
||||
import java.util.List; |
||||
|
||||
import org.springframework.stereotype.Repository; |
||||
|
||||
@Repository |
||||
public class CommandDaoImpl extends BaseDao<Command, CommandMapper> implements CommandDao { |
||||
|
||||
public CommandDaoImpl(CommandMapper commandMapper) { |
||||
super(commandMapper); |
||||
} |
||||
|
||||
@Override |
||||
public List<Command> queryCommandByIdSlot(int currentSlotIndex, int totalSlot, int idStep, int fetchNum) { |
||||
return mybatisMapper.queryCommandByIdSlot(currentSlotIndex, totalSlot, idStep, fetchNum); |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,88 @@
|
||||
/* |
||||
* 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.dolphinscheduler.dao.repository.impl; |
||||
|
||||
import static com.google.common.truth.Truth.assertThat; |
||||
|
||||
import org.apache.dolphinscheduler.common.constants.Constants; |
||||
import org.apache.dolphinscheduler.common.enums.CommandType; |
||||
import org.apache.dolphinscheduler.common.enums.FailureStrategy; |
||||
import org.apache.dolphinscheduler.common.enums.Priority; |
||||
import org.apache.dolphinscheduler.common.enums.TaskDependType; |
||||
import org.apache.dolphinscheduler.common.enums.WarningType; |
||||
import org.apache.dolphinscheduler.common.utils.DateUtils; |
||||
import org.apache.dolphinscheduler.dao.BaseDaoTest; |
||||
import org.apache.dolphinscheduler.dao.entity.Command; |
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao; |
||||
|
||||
import org.apache.commons.lang3.RandomUtils; |
||||
|
||||
import java.util.List; |
||||
|
||||
import org.junit.jupiter.api.Test; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
||||
class CommandDaoImplTest extends BaseDaoTest { |
||||
|
||||
@Autowired |
||||
private CommandDao commandDao; |
||||
|
||||
@Test |
||||
void fetchCommandByIdSlot() { |
||||
int commandSize = RandomUtils.nextInt(1, 1000); |
||||
for (int i = 0; i < commandSize; i++) { |
||||
createCommand(CommandType.START_PROCESS, 0); |
||||
} |
||||
int totalSlot = RandomUtils.nextInt(1, 10); |
||||
int currentSlotIndex = RandomUtils.nextInt(0, totalSlot); |
||||
int fetchSize = RandomUtils.nextInt(10, 100); |
||||
for (int i = 1; i < 5; i++) { |
||||
int idStep = i; |
||||
List<Command> commands = commandDao.queryCommandByIdSlot(currentSlotIndex, totalSlot, idStep, fetchSize); |
||||
assertThat(commands.size()).isGreaterThan(0); |
||||
assertThat(commands.size()) |
||||
.isEqualTo(commandDao.queryAll() |
||||
.stream() |
||||
.filter(command -> (command.getId() / idStep) % totalSlot == currentSlotIndex) |
||||
.limit(fetchSize) |
||||
.count()); |
||||
|
||||
} |
||||
|
||||
} |
||||
|
||||
private void createCommand(CommandType commandType, int processDefinitionCode) { |
||||
Command command = new Command(); |
||||
command.setCommandType(commandType); |
||||
command.setProcessDefinitionCode(processDefinitionCode); |
||||
command.setExecutorId(4); |
||||
command.setCommandParam("test command param"); |
||||
command.setTaskDependType(TaskDependType.TASK_ONLY); |
||||
command.setFailureStrategy(FailureStrategy.CONTINUE); |
||||
command.setWarningType(WarningType.ALL); |
||||
command.setWarningGroupId(1); |
||||
command.setScheduleTime(DateUtils.stringToDate("2019-12-29 12:10:00")); |
||||
command.setProcessInstancePriority(Priority.MEDIUM); |
||||
command.setStartTime(DateUtils.stringToDate("2019-12-29 10:10:00")); |
||||
command.setUpdateTime(DateUtils.stringToDate("2019-12-29 10:10:00")); |
||||
command.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP); |
||||
command.setProcessInstanceId(0); |
||||
command.setProcessDefinitionVersion(0); |
||||
commandDao.insert(command); |
||||
} |
||||
} |
||||
@ -0,0 +1,49 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.command; |
||||
|
||||
import static com.google.common.base.Preconditions.checkNotNull; |
||||
|
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao; |
||||
import org.apache.dolphinscheduler.server.master.config.CommandFetchStrategy; |
||||
import org.apache.dolphinscheduler.server.master.config.MasterConfig; |
||||
import org.apache.dolphinscheduler.server.master.registry.MasterSlotManager; |
||||
|
||||
import org.springframework.context.annotation.Bean; |
||||
import org.springframework.context.annotation.Configuration; |
||||
|
||||
@Configuration |
||||
public class CommandFetcherConfiguration { |
||||
|
||||
@Bean |
||||
public ICommandFetcher commandFetcher(MasterConfig masterConfig, |
||||
MasterSlotManager masterSlotManager, |
||||
CommandDao commandDao) { |
||||
CommandFetchStrategy commandFetchStrategy = |
||||
checkNotNull(masterConfig.getCommandFetchStrategy(), "command fetch strategy is null"); |
||||
switch (commandFetchStrategy.getType()) { |
||||
case ID_SLOT_BASED: |
||||
CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig = |
||||
(CommandFetchStrategy.IdSlotBasedFetchConfig) commandFetchStrategy.getConfig(); |
||||
return new IdSlotBasedCommandFetcher(idSlotBasedFetchConfig, masterSlotManager, commandDao); |
||||
default: |
||||
throw new IllegalArgumentException( |
||||
"unsupported command fetch strategy type: " + commandFetchStrategy.getType()); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,36 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.command; |
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command; |
||||
|
||||
import java.util.List; |
||||
|
||||
/** |
||||
* The command fetcher used to fetch commands |
||||
*/ |
||||
public interface ICommandFetcher { |
||||
|
||||
/** |
||||
* Fetch commands |
||||
* |
||||
* @return command list which need to be handled |
||||
*/ |
||||
List<Command> fetchCommands(); |
||||
|
||||
} |
||||
@ -0,0 +1,73 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.command; |
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.Command; |
||||
import org.apache.dolphinscheduler.dao.repository.CommandDao; |
||||
import org.apache.dolphinscheduler.server.master.config.CommandFetchStrategy; |
||||
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics; |
||||
import org.apache.dolphinscheduler.server.master.registry.MasterSlotManager; |
||||
|
||||
import java.util.Collections; |
||||
import java.util.List; |
||||
|
||||
import lombok.extern.slf4j.Slf4j; |
||||
|
||||
/** |
||||
* The command fetcher which is fetch commands by command id and slot. |
||||
*/ |
||||
@Slf4j |
||||
public class IdSlotBasedCommandFetcher implements ICommandFetcher { |
||||
|
||||
private final CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig; |
||||
|
||||
private final CommandDao commandDao; |
||||
|
||||
private final MasterSlotManager masterSlotManager; |
||||
|
||||
public IdSlotBasedCommandFetcher(CommandFetchStrategy.IdSlotBasedFetchConfig idSlotBasedFetchConfig, |
||||
MasterSlotManager masterSlotManager, |
||||
CommandDao commandDao) { |
||||
this.idSlotBasedFetchConfig = idSlotBasedFetchConfig; |
||||
this.masterSlotManager = masterSlotManager; |
||||
this.commandDao = commandDao; |
||||
} |
||||
|
||||
@Override |
||||
public List<Command> fetchCommands() { |
||||
long scheduleStartTime = System.currentTimeMillis(); |
||||
int currentSlotIndex = masterSlotManager.getSlot(); |
||||
int totalSlot = masterSlotManager.getMasterSize(); |
||||
if (totalSlot <= 0 || currentSlotIndex < 0) { |
||||
log.warn("Slot is validated, current master slots: {}, the current slot index is {}", totalSlot, |
||||
currentSlotIndex); |
||||
return Collections.emptyList(); |
||||
} |
||||
List<Command> commands = commandDao.queryCommandByIdSlot( |
||||
currentSlotIndex, |
||||
totalSlot, |
||||
idSlotBasedFetchConfig.getIdStep(), |
||||
idSlotBasedFetchConfig.getFetchSize()); |
||||
long cost = System.currentTimeMillis() - scheduleStartTime; |
||||
log.info("Fetch commands: {} success, cost: {}ms, totalSlot: {}, currentSlotIndex: {}", commands.size(), cost, |
||||
totalSlot, currentSlotIndex); |
||||
ProcessInstanceMetrics.recordCommandQueryTime(cost); |
||||
return commands; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,63 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.config; |
||||
|
||||
import lombok.Data; |
||||
|
||||
import org.springframework.validation.Errors; |
||||
|
||||
@Data |
||||
public class CommandFetchStrategy { |
||||
|
||||
private CommandFetchStrategyType type = CommandFetchStrategyType.ID_SLOT_BASED; |
||||
|
||||
private CommandFetchConfig config = new IdSlotBasedFetchConfig(); |
||||
|
||||
public void validate(Errors errors) { |
||||
config.validate(errors); |
||||
} |
||||
|
||||
public enum CommandFetchStrategyType { |
||||
ID_SLOT_BASED, |
||||
; |
||||
} |
||||
|
||||
public interface CommandFetchConfig { |
||||
|
||||
void validate(Errors errors); |
||||
|
||||
} |
||||
|
||||
@Data |
||||
public static class IdSlotBasedFetchConfig implements CommandFetchConfig { |
||||
|
||||
private int idStep = 1; |
||||
private int fetchSize = 10; |
||||
|
||||
@Override |
||||
public void validate(Errors errors) { |
||||
if (idStep <= 0) { |
||||
errors.rejectValue("step", null, "step must be greater than 0"); |
||||
} |
||||
if (fetchSize <= 0) { |
||||
errors.rejectValue("fetchSize", null, "fetchSize must be greater than 0"); |
||||
} |
||||
} |
||||
} |
||||
|
||||
} |
||||
Loading…
Reference in new issue