natas_hw
/
dolphinscheduler0701
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[DSIP-52][Stage #1]split tenant config from worker config

This commit is contained in:
pegasas 2024-06-28 01:51:49 -04:00
parent 225e969d2d
commit dba02afc2b
8 changed files with 26 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerTaskExecutor.java
View File

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.worker.config.TenantConfig;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
@ -32,12 +33,12 @@ import lombok.NonNull;
public class DefaultWorkerTaskExecutor extends WorkerTaskExecutor {
public DefaultWorkerTaskExecutor(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull TenantConfig tenantConfig,
@NonNull WorkerMessageSender workerMessageSender,
@Nullable StorageOperator storageOperator,
@NonNull WorkerRegistryClient workerRegistryClient) {
super(taskExecutionContext,
workerConfig,
tenantConfig,
workerMessageSender,
storageOperator,
workerRegistryClient);

9
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerTaskExecutorFactory.java
View File

@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.server.worker.runner;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.worker.config.TenantConfig;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
@ -32,18 +33,18 @@ public class DefaultWorkerTaskExecutorFactory
WorkerTaskExecutorFactory<DefaultWorkerTaskExecutor> {
private final @NonNull TaskExecutionContext taskExecutionContext;
private final @NonNull WorkerConfig workerConfig;
private final @NonNull TenantConfig tenantConfig;
private final @NonNull WorkerMessageSender workerMessageSender;
private final @Nullable StorageOperator storageOperator;
private final @NonNull WorkerRegistryClient workerRegistryClient;
public DefaultWorkerTaskExecutorFactory(@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull TenantConfig tenantConfig,
@NonNull WorkerMessageSender workerMessageSender,
@Nullable StorageOperator storageOperator,
@NonNull WorkerRegistryClient workerRegistryClient) {
this.taskExecutionContext = taskExecutionContext;
this.workerConfig = workerConfig;
this.tenantConfig = tenantConfig;
this.workerMessageSender = workerMessageSender;
this.storageOperator = storageOperator;
this.workerRegistryClient = workerRegistryClient;
@ -53,7 +54,7 @@ public class DefaultWorkerTaskExecutorFactory
public DefaultWorkerTaskExecutor createWorkerTaskExecutor() {
return new DefaultWorkerTaskExecutor(
taskExecutionContext,
workerConfig,
tenantConfig,
workerMessageSender,
storageOperator,
workerRegistryClient);

12
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutor.java
View File

@ -48,6 +48,7 @@ import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.plugin.task.api.utils.ProcessUtils;
import org.apache.dolphinscheduler.server.worker.config.TenantConfig;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
@ -74,7 +75,7 @@ public abstract class WorkerTaskExecutor implements Runnable {
protected static final Logger log = LoggerFactory.getLogger(WorkerTaskExecutor.class);
protected final TaskExecutionContext taskExecutionContext;
protected final WorkerConfig workerConfig;
protected final TenantConfig tenantConfig;
protected final WorkerMessageSender workerMessageSender;
protected final @Nullable StorageOperator storageOperator;
protected final WorkerRegistryClient workerRegistryClient;
@ -83,12 +84,12 @@ public abstract class WorkerTaskExecutor implements Runnable {
protected WorkerTaskExecutor(
@NonNull TaskExecutionContext taskExecutionContext,
@NonNull WorkerConfig workerConfig,
@NonNull TenantConfig tenantConfig,
@NonNull WorkerMessageSender workerMessageSender,
@Nullable StorageOperator storageOperator,
@NonNull WorkerRegistryClient workerRegistryClient) {
this.taskExecutionContext = taskExecutionContext;
this.workerConfig = workerConfig;
this.tenantConfig = tenantConfig;
this.workerMessageSender = workerMessageSender;
this.storageOperator = storageOperator;
this.workerRegistryClient = workerRegistryClient;
@ -207,10 +208,7 @@ public abstract class WorkerTaskExecutor implements Runnable {
log.info("Send task status {} master: {}", TaskExecutionStatus.RUNNING_EXECUTION.name(),
taskExecutionContext.getHost());
// In most of case the origin tenant is the same as the current tenant
// Except `default` tenant. The originTenant is used to download the resources
String originTenant = taskExecutionContext.getTenantCode();
taskExecutionContext.setTenantCode(TenantUtils.getOrCreateActualTenant(workerConfig, taskExecutionContext));
taskExecutionContext.setTenantCode(TenantUtils.getOrCreateActualTenant(tenantConfig, taskExecutionContext));
log.info("TenantCode: {} check successfully", taskExecutionContext.getTenantCode());
TaskExecutionContextUtils.createTaskInstanceWorkingDirectory(taskExecutionContext);

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorFactoryBuilder.java
View File

@ -55,7 +55,7 @@ public class WorkerTaskExecutorFactoryBuilder {
public WorkerTaskExecutorFactory<? extends WorkerTaskExecutor> createWorkerTaskExecutorFactory(TaskExecutionContext taskExecutionContext) {
return new DefaultWorkerTaskExecutorFactory(taskExecutionContext,
workerConfig,
workerConfig.getTenantConfig(),
workerMessageSender,
storageOperator,
workerRegistryClient);

4
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TenantUtils.java
View File

@ -46,9 +46,7 @@ public class TenantUtils {
* <p>
* If sudo is not enabled, will not check the tenant code.
*/
public static String getOrCreateActualTenant(WorkerConfig workerConfig, TaskExecutionContext taskExecutionContext) {
TenantConfig tenantConfig = workerConfig.getTenantConfig();
public static String getOrCreateActualTenant(TenantConfig tenantConfig, TaskExecutionContext taskExecutionContext) {
if (!isTenantEnable()) {
log.info("Tenant is not enabled, will use the bootstrap: {} user as tenant", getBootstrapTenant());
return getBootstrapTenant();

7
dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/DefaultWorkerTaskExecutorTest.java
View File

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.worker.config.TenantConfig;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
@ -33,7 +34,7 @@ public class DefaultWorkerTaskExecutorTest {
private TaskExecutionContext taskExecutionContext = Mockito.mock(TaskExecutionContext.class);
private WorkerConfig workerConfig = Mockito.mock(WorkerConfig.class);
private TenantConfig tenantConfig = Mockito.mock(TenantConfig.class);
private String masterAddress = "localhost:5678";
@ -53,7 +54,7 @@ public class DefaultWorkerTaskExecutorTest {
.build();
WorkerTaskExecutor workerTaskExecutor = new DefaultWorkerTaskExecutor(
taskExecutionContext,
workerConfig,
tenantConfig,
workerMessageSender,
storageOperator,
workerRegistryClient);
@ -76,7 +77,7 @@ public class DefaultWorkerTaskExecutorTest {
.build();
WorkerTaskExecutor workerTaskExecutor = new DefaultWorkerTaskExecutor(
taskExecutionContext,
workerConfig,
tenantConfig,
workerMessageSender,
storageOperator,
workerRegistryClient);

3
dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecutorThreadPoolTest.java
View File

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.worker.config.TaskExecuteThreadsFullPolicy;
import org.apache.dolphinscheduler.server.worker.config.TenantConfig;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
import org.apache.dolphinscheduler.server.worker.rpc.WorkerMessageSender;
@ -146,7 +147,7 @@ class WorkerTaskExecutorThreadPoolTest {
protected MockWorkerTaskExecutor(Runnable runnable) {
super(TaskExecutionContext.builder().taskInstanceId((int) System.nanoTime()).build(),
new WorkerConfig(),
new TenantConfig(),
new WorkerMessageSender(),
null,
new WorkerRegistryClient());

5
dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/runner/operator/TaskInstanceOperationFunctionTest.java
View File

@ -35,6 +35,7 @@ import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.worker.config.TenantConfig;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.message.MessageRetryRunner;
import org.apache.dolphinscheduler.server.worker.registry.WorkerRegistryClient;
@ -58,6 +59,8 @@ public class TaskInstanceOperationFunctionTest {
private WorkerConfig workerConfig = Mockito.mock(WorkerConfig.class);
private TenantConfig tenantConfig = Mockito.mock(TenantConfig.class);
private TaskExecutionContext taskExecutionContext = Mockito.mock(TaskExecutionContext.class);
private WorkerTaskExecutorThreadPool workerTaskExecutorThreadPool =
@ -182,6 +185,8 @@ public class TaskInstanceOperationFunctionTest {
@Test
public void testTaskInstanceDispatchOperationFunction() {
given(workerConfig.getTenantConfig()).willReturn(tenantConfig);
WorkerTaskExecutorFactoryBuilder workerTaskExecutorFactoryBuilder = new WorkerTaskExecutorFactoryBuilder(
workerConfig,
workerMessageSender,