+ * Licensed 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 com.dangdang.ddframe.job.cloud.api; + +import com.dangdang.ddframe.job.cloud.executor.TaskExecutor; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import org.apache.mesos.MesosExecutorDriver; +import org.apache.mesos.Protos; + +/** + * 云作业启动器. + * + *需将应用打包, 并在main方法中直接调用Bootstrap.execute方法
+ * + * @author caohao + * @author zhangliang + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class JobBootstrap { + + /** + * 执行作业. + */ + public static void execute() { + MesosExecutorDriver driver = new MesosExecutorDriver(new TaskExecutor()); + System.exit(Protos.Status.DRIVER_STOPPED == driver.run() ? 0 : -1); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/CloudJobFacade.java b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/CloudJobFacade.java new file mode 100644 index 0000000000..3f8d76fc2a --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/CloudJobFacade.java @@ -0,0 +1,121 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.config.JobRootConfiguration; +import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.event.JobEventBus; +import com.dangdang.ddframe.job.event.type.JobExecutionEvent; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent.Source; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent.State; +import com.dangdang.ddframe.job.exception.JobExecutionEnvironmentException; +import com.dangdang.ddframe.job.executor.JobFacade; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import lombok.RequiredArgsConstructor; + +import java.util.Collection; + +/** + * 为调度器提供内部服务的门面类. + * + * @author zhangliang + */ +@RequiredArgsConstructor +public class CloudJobFacade implements JobFacade { + + private final ShardingContexts shardingContexts; + + private final JobConfigurationContext jobConfig; + + private final JobEventBus jobEventBus; + + @Override + public JobRootConfiguration loadJobRootConfiguration(final boolean fromCache) { + return jobConfig; + } + + @Override + public void checkJobExecutionEnvironment() throws JobExecutionEnvironmentException { + } + + @Override + public void failoverIfNecessary() { + } + + @Override + public void registerJobBegin(final ShardingContexts shardingContexts) { + } + + @Override + public void registerJobCompleted(final ShardingContexts shardingContexts) { + } + + public ShardingContexts getShardingContexts() { + return shardingContexts; + } + + @Override + public boolean misfireIfNecessary(final Collection+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.api.ElasticJob; +import com.dangdang.ddframe.job.config.JobRootConfiguration; +import com.dangdang.ddframe.job.exception.JobSystemException; +import com.dangdang.ddframe.job.executor.JobExecutorFactory; +import com.dangdang.ddframe.job.executor.JobFacade; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import lombok.RequiredArgsConstructor; +import lombok.Setter; +import org.apache.mesos.ExecutorDriver; +import org.apache.mesos.Protos; +import org.quartz.CronScheduleBuilder; +import org.quartz.CronTrigger; +import org.quartz.Job; +import org.quartz.JobBuilder; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.quartz.Scheduler; +import org.quartz.SchedulerException; +import org.quartz.TriggerBuilder; +import org.quartz.impl.StdSchedulerFactory; +import org.quartz.plugins.management.ShutdownHookPlugin; + +import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; + +/** + * 常驻作业调度器. + * + * @author zhangliang + * @author caohao + */ +@RequiredArgsConstructor +public final class DaemonTaskScheduler { + + public static final String ELASTIC_JOB_DATA_MAP_KEY = "elasticJob"; + + private static final String JOB_FACADE_DATA_MAP_KEY = "jobFacade"; + + private static final String EXECUTOR_DRIVER_DATA_MAP_KEY = "executorDriver"; + + private static final String TASK_ID_DATA_MAP_KEY = "taskId"; + + private static final ConcurrentHashMap+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.api.JobType; +import com.dangdang.ddframe.job.config.JobCoreConfiguration; +import com.dangdang.ddframe.job.config.JobRootConfiguration; +import com.dangdang.ddframe.job.config.JobTypeConfiguration; +import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration; +import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration; +import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration; +import com.dangdang.ddframe.job.executor.handler.JobProperties.JobPropertiesEnum; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import lombok.Getter; + +import java.util.Map; + +/** + * 内部的作业配置上下文. + * + * @author caohao + */ +public class JobConfigurationContext implements JobRootConfiguration { + + private static final String IGNORE_CRON = "ignoredCron"; + + private JobTypeConfiguration jobTypeConfig; + + @Getter + private String beanName; + + @Getter + private String applicationContext; + + public JobConfigurationContext(final Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.api.ElasticJob; +import com.dangdang.ddframe.job.api.script.ScriptJob; +import com.dangdang.ddframe.job.event.JobEventBus; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration; +import com.dangdang.ddframe.job.exception.JobSystemException; +import com.dangdang.ddframe.job.executor.JobExecutorFactory; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import com.dangdang.ddframe.job.util.concurrent.ExecutorServiceObject; +import com.google.common.base.Strings; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.dbcp.BasicDataSource; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.mesos.Executor; +import org.apache.mesos.ExecutorDriver; +import org.apache.mesos.Protos; +import org.apache.mesos.Protos.TaskInfo; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ExecutorService; + +/** + * 作业任务执行器. + * + * @author zhangliang + */ +@Slf4j +public final class TaskExecutor implements Executor { + + private final ExecutorService executorService; + + private final Map* Licensed 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 - * + * + * 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. @@ -15,22 +15,22 @@ *
*/ -package com.dangdang.ddframe.reg; +package com.dangdang.ddframe.job.cloud.executor; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; import org.junit.runner.RunWith; import org.junit.runners.Suite; import org.junit.runners.Suite.SuiteClasses; -import com.dangdang.ddframe.reg.exception.RegExceptionHandlerTest; -import com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenterForAuthTest; -import com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenterTest; - @RunWith(Suite.class) @SuiteClasses({ - ZookeeperRegistryCenterTest.class, - ZookeeperRegistryCenterForAuthTest.class, - RegExceptionHandlerTest.class + CloudJobFacadeTest.class, + DaemonTaskSchedulerTest.class, + JobConfigurationContextTest.class, + TaskExecutorTest.class, + TaskExecutorThreadTest.class }) -public class AllRegTests { - +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class AllCloudExecutorTests { } diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/CloudJobFacadeTest.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/CloudJobFacadeTest.java new file mode 100644 index 0000000000..92d980d847 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/CloudJobFacadeTest.java @@ -0,0 +1,168 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.api.ElasticJob; +import com.dangdang.ddframe.job.api.JobType; +import com.dangdang.ddframe.job.config.JobRootConfiguration; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.event.JobEventBus; +import com.dangdang.ddframe.job.event.type.JobExecutionEvent; +import com.dangdang.ddframe.job.event.type.JobExecutionEvent.ExecutionSource; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent.State; +import com.dangdang.ddframe.job.exception.JobExecutionEnvironmentException; +import com.dangdang.ddframe.job.executor.JobFacade; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import java.util.HashMap; +import java.util.Map; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.verify; + +public class CloudJobFacadeTest { + + private final ShardingContexts shardingContexts; + + private final JobConfigurationContext jobConfig; + + @Mock + private JobEventBus eventBus; + + private final JobFacade jobFacade; + + public CloudJobFacadeTest() { + MockitoAnnotations.initMocks(this); + shardingContexts = getShardingContexts(); + jobConfig = new JobConfigurationContext(getJobConfigurationMap(JobType.SIMPLE, false)); + jobFacade = new CloudJobFacade(shardingContexts, jobConfig, eventBus); + } + + private ShardingContexts getShardingContexts() { + Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.cloud.executor.fixture.TestScriptJobConfiguration; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.executor.AbstractElasticJobExecutor; +import com.dangdang.ddframe.job.executor.JobFacade; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import org.apache.mesos.ExecutorDriver; +import org.apache.mesos.Protos.TaskID; +import org.apache.mesos.Protos.TaskState; +import org.apache.mesos.Protos.TaskStatus; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.quartz.JobExecutionContext; + +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public final class DaemonTaskSchedulerTest { + + @Mock + private JobFacade jobFacade; + + @Mock + private ExecutorDriver executorDriver; + + @Mock + private JobExecutionContext jobExecutionContext; + + @Mock + private AbstractElasticJobExecutor jobExecutor; + + @Mock + private ShardingContexts shardingContexts; + + private TaskID taskId = TaskID.newBuilder().setValue(String.format("%s@-@0@-@%s@-@fake_slave_id@-@0", "test_job", ExecutionType.READY)).build(); + + private DaemonTaskScheduler.DaemonJob daemonJob; + + @Before + public void setUp() throws NoSuchFieldException { + daemonJob = new DaemonTaskScheduler.DaemonJob(); + daemonJob.setElasticJob(null); + daemonJob.setJobFacade(jobFacade); + daemonJob.setExecutorDriver(executorDriver); + daemonJob.setTaskId(taskId); + } + + @Test + public void assertJobRun() throws Exception { + when(jobFacade.getShardingContexts()).thenReturn(shardingContexts); + when(jobFacade.loadJobRootConfiguration(true)).thenReturn(new TestScriptJobConfiguration("test.sh")); + daemonJob.execute(jobExecutionContext); + verify(shardingContexts).setAllowSendJobEvent(true); + verify(executorDriver).sendStatusUpdate(TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_RUNNING).setMessage("BEGIN").build()); + verify(executorDriver).sendStatusUpdate(TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_RUNNING).setMessage("COMPLETE").build()); + verify(shardingContexts).setCurrentJobEventSamplingCount(0); + } + + @Test + public void assertJobRunWithEventSampling() throws Exception { + when(shardingContexts.getJobEventSamplingCount()).thenReturn(2); + when(jobFacade.getShardingContexts()).thenReturn(shardingContexts); + when(jobFacade.loadJobRootConfiguration(true)).thenReturn(new TestScriptJobConfiguration("test.sh")); + daemonJob.execute(jobExecutionContext); + verify(shardingContexts).setCurrentJobEventSamplingCount(1); + verify(shardingContexts).setAllowSendJobEvent(false); + when(shardingContexts.getCurrentJobEventSamplingCount()).thenReturn(1); + daemonJob.execute(jobExecutionContext); + verify(shardingContexts).setAllowSendJobEvent(true); + verify(executorDriver).sendStatusUpdate(TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_RUNNING).setMessage("BEGIN").build()); + verify(executorDriver).sendStatusUpdate(TaskStatus.newBuilder().setTaskId(taskId).setState(TaskState.TASK_RUNNING).setMessage("COMPLETE").build()); + verify(shardingContexts).setCurrentJobEventSamplingCount(0); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/JobConfigurationContextTest.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/JobConfigurationContextTest.java new file mode 100644 index 0000000000..d1475de933 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/JobConfigurationContextTest.java @@ -0,0 +1,82 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.api.JobType; +import com.dangdang.ddframe.job.cloud.executor.fixture.TestJob; +import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration; +import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration; +import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration; +import com.dangdang.ddframe.job.exception.JobExecutionEnvironmentException; +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; + +import static junit.framework.TestCase.assertFalse; +import static org.hamcrest.core.Is.is; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +public class JobConfigurationContextTest { + + @Test + public void assertSimpleJobConfigurationContext() throws JobExecutionEnvironmentException { + assertTrue(new JobConfigurationContext(buildJobConfigurationContextMap(JobType.SIMPLE)).getTypeConfig() instanceof SimpleJobConfiguration); + } + + @Test + public void assertDataflowJobConfigurationContext() throws JobExecutionEnvironmentException { + assertTrue(new JobConfigurationContext(buildJobConfigurationContextMap(JobType.DATAFLOW)).getTypeConfig() instanceof DataflowJobConfiguration); + } + + @Test + public void assertScriptJobConfigurationContext() throws JobExecutionEnvironmentException { + assertTrue(new JobConfigurationContext(buildJobConfigurationContextMap(JobType.SCRIPT)).getTypeConfig() instanceof ScriptJobConfiguration); + } + + @Test + public void assertSpringSimpleJobConfigurationContext() throws JobExecutionEnvironmentException { + Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.google.protobuf.ByteString; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.mesos.ExecutorDriver; +import org.apache.mesos.Protos; +import org.apache.mesos.Protos.ExecutorInfo; +import org.apache.mesos.Protos.FrameworkInfo; +import org.apache.mesos.Protos.SlaveInfo; +import org.apache.mesos.Protos.TaskID; +import org.apache.mesos.Protos.TaskInfo; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.unitils.util.ReflectionUtils; + +import java.util.HashMap; +import java.util.concurrent.ExecutorService; + +import static org.mockito.Mockito.verify; + +@RunWith(MockitoJUnitRunner.class) +public final class TaskExecutorTest { + + @Mock + private ExecutorDriver executorDriver; + + @Mock + private ExecutorService executorService; + + private ExecutorInfo executorInfo; + + private SlaveInfo slaveInfo = SlaveInfo.getDefaultInstance(); + + private FrameworkInfo frameworkInfo = FrameworkInfo.getDefaultInstance(); + + private TaskExecutor taskExecutor; + + @Before + public void setUp() throws NoSuchFieldException { + taskExecutor = new TaskExecutor(); + ReflectionUtils.setFieldValue(taskExecutor, "executorService", executorService); + executorInfo = ExecutorInfo.getDefaultInstance(); + } + + @Test + public void assertKillTask() { + TaskID taskID = Protos.TaskID.newBuilder().setValue("task_id").build(); + taskExecutor.killTask(executorDriver, taskID); + verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskID).setState(Protos.TaskState.TASK_KILLED).build()); + } + + @Test + public void assertRegisteredWithoutData() { + // CHECKSTYLE:OFF + HashMap+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor; + +import com.dangdang.ddframe.job.api.JobType; +import com.dangdang.ddframe.job.cloud.executor.TaskExecutor.TaskThread; +import com.dangdang.ddframe.job.cloud.executor.fixture.TestJob; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.exception.JobSystemException; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import com.google.protobuf.ByteString; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.mesos.ExecutorDriver; +import org.apache.mesos.Protos; +import org.apache.mesos.Protos.TaskID; +import org.apache.mesos.Protos.TaskInfo; +import org.apache.mesos.Protos.TaskState; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + +import static junit.framework.TestCase.assertTrue; +import static org.mockito.Mockito.verify; + +@RunWith(MockitoJUnitRunner.class) +public final class TaskExecutorThreadTest { + + @Mock + private ExecutorDriver executorDriver; + + private final String taskId = String.format("%s@-@0@-@%s@-@fake_slave_id@-@0", "test_job", ExecutionType.READY); + + @Test + public void assertLaunchTaskWithDaemonTaskAndJavaSimpleJob() { + TaskInfo taskInfo = buildJavaTransientTaskInfo(); + TaskThread taskThread = new TaskExecutor().new TaskThread(executorDriver, taskInfo); + taskThread.run(); + verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_RUNNING).build()); + verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(Protos.TaskState.TASK_FINISHED).build()); + } + + @Test + public void assertLaunchTaskWithTransientTaskAndSpringSimpleJob() { + TaskInfo taskInfo = buildSpringDaemonTaskInfo(); + TaskThread taskThread = new TaskExecutor().new TaskThread(executorDriver, taskInfo); + taskThread.run(); + verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_RUNNING).build()); + } + + @Test + public void assertLaunchTaskWithDaemonTaskAndJavaScriptJob() { + TaskInfo taskInfo = buildSpringScriptTransientTaskInfo(); + TaskThread taskThread = new TaskExecutor().new TaskThread(executorDriver, taskInfo); + taskThread.run(); + verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(TaskState.TASK_RUNNING).build()); + verify(executorDriver).sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(Protos.TaskState.TASK_FINISHED).build()); + } + + @Test + public void assertLaunchTaskWithWrongElasticJobClass() { + TaskInfo taskInfo = buildWrongElasticJobClass(); + TaskThread taskThread = new TaskExecutor().new TaskThread(executorDriver, taskInfo); + try { + taskThread.run(); + } catch (final JobSystemException ex) { + assertTrue(ex.getMessage().startsWith("Elastic-Job: Class 'com.dangdang.ddframe.job.cloud.executor.TaskExecutorThreadTest' must implements ElasticJob interface.")); + } + } + + @Test + public void assertLaunchTaskWithWrongClass() { + TaskInfo taskInfo = buildWrongClass(); + TaskThread taskThread = new TaskExecutor().new TaskThread(executorDriver, taskInfo); + try { + taskThread.run(); + } catch (final JobSystemException ex) { + assertTrue(ex.getMessage().startsWith("Elastic-Job: Class 'WrongClass' initialize failure, the error message is 'WrongClass'.")); + } + } + + private TaskInfo buildWrongClass() { + return buildTaskInfo(buildBaseJobConfigurationContextMapWithJobClassAndCron("WrongClass", "ignoredCron")).build(); + } + + private TaskInfo buildWrongElasticJobClass() { + return buildTaskInfo(buildBaseJobConfigurationContextMapWithJobClassAndCron(TaskExecutorThreadTest.class.getCanonicalName(), "ignoredCron")).build(); + } + + private TaskInfo buildSpringDaemonTaskInfo() { + return buildTaskInfo(buildSpringJobConfigurationContextMap()).build(); + } + + private TaskInfo buildJavaTransientTaskInfo() { + return buildTaskInfo(buildBaseJobConfigurationContextMapWithJobClassAndCron(TestJob.class.getCanonicalName(), "ignoredCron")).build(); + } + + private TaskInfo buildSpringScriptTransientTaskInfo() { + return buildTaskInfo(buildBaseJobConfigurationContextMap(TestJob.class.getCanonicalName(), "ignoredCron", JobType.SCRIPT)).build(); + } + + private TaskInfo.Builder buildTaskInfo(final Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor.fixture; + +import com.dangdang.ddframe.job.api.ShardingContext; +import com.dangdang.ddframe.job.api.simple.SimpleJob; + +public final class TestJob implements SimpleJob { + + @Override + public void execute(final ShardingContext shardingContext) { + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/fixture/TestScriptJobConfiguration.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/fixture/TestScriptJobConfiguration.java new file mode 100644 index 0000000000..0114a4a175 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/fixture/TestScriptJobConfiguration.java @@ -0,0 +1,38 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.executor.fixture; + + +import com.dangdang.ddframe.job.config.JobCoreConfiguration; +import com.dangdang.ddframe.job.config.JobRootConfiguration; +import com.dangdang.ddframe.job.config.JobTypeConfiguration; +import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration; +import com.dangdang.ddframe.job.executor.handler.JobProperties; +import lombok.RequiredArgsConstructor; + +@RequiredArgsConstructor +public final class TestScriptJobConfiguration implements JobRootConfiguration { + + private final String scriptCommandLine; + + @Override + public JobTypeConfiguration getTypeConfig() { + return new ScriptJobConfiguration(JobCoreConfiguration.newBuilder("test_script_job", "0/1 * * * * ?", 3) + .jobProperties(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), "ignoredExceptionHandler").build(), scriptCommandLine); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml b/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml new file mode 100644 index 0000000000..033830355a --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml @@ -0,0 +1,8 @@ + ++ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler; + +import com.dangdang.ddframe.job.cloud.scheduler.env.BootstrapEnvironment; +import com.dangdang.ddframe.job.cloud.scheduler.ha.HANode; +import com.dangdang.ddframe.job.cloud.scheduler.ha.SchedulerElectionCandidate; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperElectionService; +import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter; +import lombok.extern.slf4j.Slf4j; +import org.apache.curator.framework.CuratorFramework; + +import java.util.concurrent.CountDownLatch; + +/** + * Mesos框架启动器. + * + * @author caohao + */ +@Slf4j +public final class Bootstrap { + + /** + * 启动入口. + * + * @param args 命令行参数无需传入 + * @throws InterruptedException 线程中断异常 + */ + // CHECKSTYLE:OFF + public static void main(final String[] args) throws InterruptedException { + // CHECKSTYLE:ON + CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(BootstrapEnvironment.getInstance().getZookeeperConfiguration()); + regCenter.init(); + final ZookeeperElectionService electionService = new ZookeeperElectionService( + BootstrapEnvironment.getInstance().getFrameworkHostPort(), (CuratorFramework) regCenter.getRawClient(), HANode.ELECTION_NODE, new SchedulerElectionCandidate(regCenter)); + electionService.start(); + final CountDownLatch latch = new CountDownLatch(1); + latch.await(); + Runtime.getRuntime().addShutdownHook(new Thread("shutdown-hook") { + + @Override + public void run() { + electionService.stop(); + latch.countDown(); + } + }); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfiguration.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfiguration.java new file mode 100644 index 0000000000..f77cff2afc --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfiguration.java @@ -0,0 +1,49 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.app; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.ToString; + +/** + * 云作业App配置对象. + * + * @author caohao + */ +@AllArgsConstructor +@RequiredArgsConstructor +@Getter +@ToString +public final class CloudAppConfiguration { + + private final String appName; + + private final String appURL; + + private final String bootstrapScript; + + private double cpuCount = 1; + + private double memoryMB = 128; + + private boolean appCacheEnable = true; + + private int eventTraceSamplingCount; +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfigurationGsonFactory.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfigurationGsonFactory.java new file mode 100644 index 0000000000..63e4391b72 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfigurationGsonFactory.java @@ -0,0 +1,124 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.app; + +import com.dangdang.ddframe.job.util.json.GsonFactory; +import com.google.gson.TypeAdapter; +import com.google.gson.stream.JsonReader; +import com.google.gson.stream.JsonWriter; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +import java.io.IOException; + +/** + * 云作业App配置的Gson工厂. + * + * @author caohao + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class CloudAppConfigurationGsonFactory { + + static { + GsonFactory.registerTypeAdapter(CloudAppConfiguration.class, new CloudAppConfigurationGsonTypeAdapter()); + } + + /** + * 将云作业App配置转换为JSON字符串. + * + * @param cloudAppConfig 云作业App配置对象 + * @return 作业配置JSON字符串 + */ + public static String toJson(final CloudAppConfiguration cloudAppConfig) { + return GsonFactory.getGson().toJson(cloudAppConfig); + } + + /** + * 将JSON字符串转换为云作业App配置. + * + * @param cloudAppConfigJson 云作业App配置JSON字符串 + * @return 作业配置对象 + */ + public static CloudAppConfiguration fromJson(final String cloudAppConfigJson) { + return GsonFactory.getGson().fromJson(cloudAppConfigJson, CloudAppConfiguration.class); + } + + /** + * 云作业App配置的Json转换适配器. + * + * @author caohao + */ + public static final class CloudAppConfigurationGsonTypeAdapter extends TypeAdapter+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.app; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +/** + * 云作业App配置节点路径. + * + * @author caohao + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class CloudAppConfigurationNode { + + public static final String ROOT = "/config/app"; + + private static final String APP_CONFIG = ROOT + "/%s"; + + static String getRootNodePath(final String appName) { + return String.format(APP_CONFIG, appName); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfigurationService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfigurationService.java new file mode 100644 index 0000000000..d946311e62 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/app/CloudAppConfigurationService.java @@ -0,0 +1,96 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.app; + +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Optional; +import lombok.RequiredArgsConstructor; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +/** + * 云作业App配置服务. + * + * @author caohao + */ +@RequiredArgsConstructor +public class CloudAppConfigurationService { + + private final CoordinatorRegistryCenter regCenter; + + /** + * 添加云作业APP配置. + * + * @param appConfig 云作业App配置对象 + */ + public void add(final CloudAppConfiguration appConfig) { + regCenter.persist(CloudAppConfigurationNode.getRootNodePath(appConfig.getAppName()), CloudAppConfigurationGsonFactory.toJson(appConfig)); + } + + /** + * 修改云作业APP配置. + * + * @param appConfig 云作业App配置对象 + */ + public void update(final CloudAppConfiguration appConfig) { + regCenter.update(CloudAppConfigurationNode.getRootNodePath(appConfig.getAppName()), CloudAppConfigurationGsonFactory.toJson(appConfig)); + } + + /** + * 根据云作业App名称获取App配置. + * + * @param appName 云作业App名称 + * @return 云作业App配置 + */ + public Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.job; + +import com.dangdang.ddframe.job.config.JobRootConfiguration; +import com.dangdang.ddframe.job.config.JobTypeConfiguration; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * 云作业配置对象. + * + * @author zhangliang + */ +@RequiredArgsConstructor +@AllArgsConstructor +@Getter +public final class CloudJobConfiguration implements JobRootConfiguration { + + private final String appName; + + private final JobTypeConfiguration typeConfig; + + private final double cpuCount; + + private final double memoryMB; + + private final CloudJobExecutionType jobExecutionType; + + private String beanName; + + private String applicationContext; + + /** + * 获取作业名称. + * + * @return 作业名称 + */ + public String getJobName() { + return typeConfig.getCoreConfig().getJobName(); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationGsonFactory.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationGsonFactory.java new file mode 100644 index 0000000000..123f15f849 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationGsonFactory.java @@ -0,0 +1,127 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.job; + +import com.dangdang.ddframe.job.config.JobTypeConfiguration; +import com.dangdang.ddframe.job.util.json.AbstractJobConfigurationGsonTypeAdapter; +import com.dangdang.ddframe.job.util.json.GsonFactory; +import com.google.common.base.Preconditions; +import com.google.gson.stream.JsonReader; +import com.google.gson.stream.JsonWriter; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +import java.io.IOException; +import java.util.Map; + +/** + * Cloud作业配置的Gson工厂. + * + * @author zhangliang + * @author caohao + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class CloudJobConfigurationGsonFactory { + + static { + GsonFactory.registerTypeAdapter(CloudJobConfiguration.class, new CloudJobConfigurationGsonTypeAdapter()); + } + + /** + * 将作业配置转换为JSON字符串. + * + * @param cloudJobConfig 作业配置对象 + * @return 作业配置JSON字符串 + */ + public static String toJson(final CloudJobConfiguration cloudJobConfig) { + return GsonFactory.getGson().toJson(cloudJobConfig); + } + + /** + * 将JSON字符串转换为作业配置. + * + * @param cloudJobConfigJson 作业配置JSON字符串 + * @return 作业配置对象 + */ + public static CloudJobConfiguration fromJson(final String cloudJobConfigJson) { + return GsonFactory.getGson().fromJson(cloudJobConfigJson, CloudJobConfiguration.class); + } + + /** + * Cloud作业配置的Json转换适配器. + * + * @author zhangliang + */ + public static final class CloudJobConfigurationGsonTypeAdapter extends AbstractJobConfigurationGsonTypeAdapter+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.job; + +import com.dangdang.ddframe.job.cloud.scheduler.producer.ProducerManager; +import com.dangdang.ddframe.job.cloud.scheduler.state.ready.ReadyService; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.cache.TreeCache; +import org.apache.curator.framework.recipes.cache.TreeCacheEvent; +import org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type; +import org.apache.curator.framework.recipes.cache.TreeCacheListener; + +import java.util.Collections; +import java.util.concurrent.Executors; + +/** + * 云作业配置变更监听. + * + * @author zhangliang + * @author caohao + */ +public class CloudJobConfigurationListener implements TreeCacheListener { + + private final CoordinatorRegistryCenter regCenter; + + private final ProducerManager producerManager; + + private final ReadyService readyService; + + public CloudJobConfigurationListener(final CoordinatorRegistryCenter regCenter, final ProducerManager producerManager) { + this.regCenter = regCenter; + readyService = new ReadyService(regCenter); + this.producerManager = producerManager; + } + + @Override + public void childEvent(final CuratorFramework client, final TreeCacheEvent event) throws Exception { + String path = null == event.getData() ? "" : event.getData().getPath(); + if (isJobConfigNode(event, path, Type.NODE_ADDED)) { + CloudJobConfiguration jobConfig = getJobConfig(event); + if (null != jobConfig) { + producerManager.schedule(jobConfig); + } + } else if (isJobConfigNode(event, path, Type.NODE_UPDATED)) { + CloudJobConfiguration jobConfig = getJobConfig(event); + if (null == jobConfig) { + return; + } + if (CloudJobExecutionType.DAEMON == jobConfig.getJobExecutionType()) { + readyService.remove(Collections.singletonList(jobConfig.getJobName())); + } + if (!jobConfig.getTypeConfig().getCoreConfig().isMisfire()) { + readyService.setMisfireDisabled(jobConfig.getJobName()); + } + producerManager.reschedule(jobConfig); + } else if (isJobConfigNode(event, path, Type.NODE_REMOVED)) { + String jobName = path.substring(CloudJobConfigurationNode.ROOT.length() + 1, path.length()); + producerManager.unschedule(jobName); + } + } + + private boolean isJobConfigNode(final TreeCacheEvent event, final String path, final Type type) { + return type == event.getType() && path.startsWith(CloudJobConfigurationNode.ROOT) && path.length() > CloudJobConfigurationNode.ROOT.length(); + } + + private CloudJobConfiguration getJobConfig(final TreeCacheEvent event) { + try { + return CloudJobConfigurationGsonFactory.fromJson(new String(event.getData().getData())); + // CHECKSTYLE:OFF + } catch (final Exception ex) { + // CHECKSTYLE:ON + return null; + } + } + + /** + * 启动云作业配置变更监听服务. + */ + public void start() { + getCache().getListenable().addListener(this, Executors.newSingleThreadExecutor()); + } + + /** + * 停止云作业配置变更监听服务. + */ + public void stop() { + getCache().getListenable().removeListener(this); + } + + private TreeCache getCache() { + TreeCache result = (TreeCache) regCenter.getRawCache(CloudJobConfigurationNode.ROOT); + if (null != result) { + return result; + } + regCenter.addCacheData(CloudJobConfigurationNode.ROOT); + return (TreeCache) regCenter.getRawCache(CloudJobConfigurationNode.ROOT); + } +} diff --git a/elastic-job-core/src/test/java/com/dangdang/ddframe/job/internal/env/FakeLocalHostService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationNode.java similarity index 52% rename from elastic-job-core/src/test/java/com/dangdang/ddframe/job/internal/env/FakeLocalHostService.java rename to elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationNode.java index 2809223e3a..82efa26031 100644 --- a/elastic-job-core/src/test/java/com/dangdang/ddframe/job/internal/env/FakeLocalHostService.java +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationNode.java @@ -1,12 +1,12 @@ -/** +/* * Copyright 1999-2015 dangdang.com. ** Licensed 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 - * + * + * 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. @@ -15,28 +15,24 @@ *
*/ -package com.dangdang.ddframe.job.internal.env; +package com.dangdang.ddframe.job.cloud.scheduler.config.job; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; /** - * 为测试使用的获取本地网络的服务. - * + * 作业配置节点路径. + * * @author zhangliang */ -public final class FakeLocalHostService implements LocalHostService { +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class CloudJobConfigurationNode { - private final String localHostIp; + public static final String ROOT = "/config/job"; - public FakeLocalHostService(final String localHostIp) { - this.localHostIp = localHostIp; - } - - @Override - public String getIp() { - return localHostIp; - } + private static final String JOB_CONFIG = ROOT + "/%s"; - @Override - public String getHostName() { - return localHostIp + "_hostMame"; + static String getRootNodePath(final String jobName) { + return String.format(JOB_CONFIG, jobName); } } diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationService.java new file mode 100644 index 0000000000..a9c638562a --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/config/job/CloudJobConfigurationService.java @@ -0,0 +1,95 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.config.job; + +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Optional; +import lombok.RequiredArgsConstructor; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +/** + * 作业配置服务. + * + * @author zhangliang + */ +@RequiredArgsConstructor +public class CloudJobConfigurationService { + + private final CoordinatorRegistryCenter regCenter; + + /** + * 添加云作业配置. + * + * @param jobConfig 云作业配置对象 + */ + public void add(final CloudJobConfiguration jobConfig) { + regCenter.persist(CloudJobConfigurationNode.getRootNodePath(jobConfig.getJobName()), CloudJobConfigurationGsonFactory.toJson(jobConfig)); + } + + /** + * 修改云作业配置. + * + * @param jobConfig 云作业配置对象 + */ + public void update(final CloudJobConfiguration jobConfig) { + regCenter.update(CloudJobConfigurationNode.getRootNodePath(jobConfig.getJobName()), CloudJobConfigurationGsonFactory.toJson(jobConfig)); + } + + /** + * 获取所有注册的云作业配置. + * + * @return 注册的云作业配置 + */ + public Collection* Licensed 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 - * + * + * 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. @@ -15,9 +15,15 @@ *
*/ -package com.dangdang.ddframe.job.console.exception; +package com.dangdang.ddframe.job.cloud.scheduler.config.job; -public final class NoRegistryCenterException extends RuntimeException { +/** + * 作业执行类型. + * + * @author zhangliang + */ +public enum CloudJobExecutionType { + + DAEMON, TRANSIENT - private static final long serialVersionUID = -7230151498491198890L; } diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/context/JobContext.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/context/JobContext.java new file mode 100644 index 0000000000..31f4154082 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/context/JobContext.java @@ -0,0 +1,58 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.context; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.context.ExecutionType; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +import java.util.ArrayList; +import java.util.List; + +/** + * 作业运行上下文. + * + * @author zhangliang + */ +@RequiredArgsConstructor +@Getter +public final class JobContext { + + private final CloudJobConfiguration jobConfig; + + private final List+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.env; + +import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration; +import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration; +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.dbcp.BasicDataSource; + +import java.io.FileInputStream; +import java.io.IOException; +import java.util.HashMap; +import java.util.Properties; + +/** + * 启动环境对象. + * + * @author zhangliang + */ +@Slf4j +public final class BootstrapEnvironment { + + @Getter + private static BootstrapEnvironment instance = new BootstrapEnvironment(); + + private static final String PROPERTIES_PATH = "conf/elastic-job-cloud-scheduler.properties"; + + private final Properties properties; + + private BootstrapEnvironment() { + properties = getProperties(); + } + + private Properties getProperties() { + Properties result = new Properties(); + try (FileInputStream fileInputStream = new FileInputStream(PROPERTIES_PATH)) { + result.load(fileInputStream); + } catch (final IOException ex) { + log.warn("Cannot found conf/elastic-job-cloud-scheduler.properties, use default value now."); + } + return result; + } + + /** + * 获取Framework的Hostname和Port. + * + * @return Framework的Hostname和Port + */ + public String getFrameworkHostPort() { + return String.format("%s:%d", getMesosConfiguration().getHostname(), getRestfulServerConfiguration().getPort()); + } + + /** + * 获取Mesos配置对象. + * + * @return Mesos配置对象 + */ + public MesosConfiguration getMesosConfiguration() { + return new MesosConfiguration(getValue(EnvironmentArgument.USER), getValue(EnvironmentArgument.MESOS_URL), getValue(EnvironmentArgument.HOSTNAME)); + } + + /** + * 获取Zookeeper配置对象. + * + * @return Zookeeper配置对象 + */ + // TODO 其他zkConfig的值可配置 + public ZookeeperConfiguration getZookeeperConfiguration() { + ZookeeperConfiguration result = new ZookeeperConfiguration(getValue(EnvironmentArgument.ZOOKEEPER_SERVERS), getValue(EnvironmentArgument.ZOOKEEPER_NAMESPACE)); + String digest = getValue(EnvironmentArgument.ZOOKEEPER_DIGEST); + if (!Strings.isNullOrEmpty(digest)) { + result.setDigest(digest); + } + return result; + } + + /** + * 获取Restful服务器配置对象. + * + * @return Restful服务器配置对象 + */ + public RestfulServerConfiguration getRestfulServerConfiguration() { + return new RestfulServerConfiguration(Integer.parseInt(getValue(EnvironmentArgument.PORT))); + } + + /** + * 获取Mesos框架配置对象. + * + * @return Mesos框架配置对象 + */ + public FrameworkConfiguration getFrameworkConfiguration() { + return new FrameworkConfiguration(Integer.parseInt(getValue(EnvironmentArgument.JOB_STATE_QUEUE_SIZE))); + } + + /** + * 获取作业数据库事件配置. + * + * @return 作业数据库事件配置 + */ + public Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.env; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * Mesos框架配置项. + * + * @author zhangliang + */ +@RequiredArgsConstructor +@Getter +public final class FrameworkConfiguration { + + private final int jobStateQueueSize; +} diff --git a/elastic-job-core/src/main/java/com/dangdang/ddframe/job/api/AbstractJobExecutionShardingContext.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/env/MesosConfiguration.java similarity index 50% rename from elastic-job-core/src/main/java/com/dangdang/ddframe/job/api/AbstractJobExecutionShardingContext.java rename to elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/env/MesosConfiguration.java index 2471b42f51..3c9de4869e 100644 --- a/elastic-job-core/src/main/java/com/dangdang/ddframe/job/api/AbstractJobExecutionShardingContext.java +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/env/MesosConfiguration.java @@ -1,12 +1,12 @@ -/** +/* * Copyright 1999-2015 dangdang.com. ** Licensed 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 - * + * + * 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. @@ -15,38 +15,33 @@ *
*/ -package com.dangdang.ddframe.job.api; +package com.dangdang.ddframe.job.cloud.scheduler.env; import lombok.Getter; -import lombok.Setter; +import lombok.RequiredArgsConstructor; +/** + * Mesos配置项. + * + * @author zhangliang + */ +@RequiredArgsConstructor @Getter -@Setter -abstract class AbstractJobExecutionShardingContext { +public final class MesosConfiguration { /** - * 作业名称. + * 框架名称. */ - private String jobName; + public static final String FRAMEWORK_NAME = "Elastic-Job-Cloud"; /** - * 分片总数. + * 框架Failover超时时间,默认为1周. */ - private int shardingTotalCount; + public static final double FRAMEWORK_FAILOVER_TIMEOUT = 60 * 60 * 24 * 7; - /** - * 作业自定义参数. - * 可以配置多个相同的作业, 但是用不同的参数作为不同的调度实例. - */ - private String jobParameter; + private final String user; - /** - * 监控作业执行时状态. - */ - private boolean monitorExecution; + private final String url; - /** - * 每次抓取的数据量. - */ - private int fetchDataCount; + private final String hostname; } diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/env/RestfulServerConfiguration.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/env/RestfulServerConfiguration.java new file mode 100644 index 0000000000..c9f0796964 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/env/RestfulServerConfiguration.java @@ -0,0 +1,33 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.env; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * Restful服务器配置项. + * + * @author zhangliang + */ +@RequiredArgsConstructor +@Getter +public final class RestfulServerConfiguration { + + private final int port; +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/ha/FrameworkIDService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/ha/FrameworkIDService.java new file mode 100644 index 0000000000..6d11e6c5d4 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/ha/FrameworkIDService.java @@ -0,0 +1,55 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.ha; + +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Optional; +import com.google.common.base.Strings; +import lombok.RequiredArgsConstructor; + +/** + * FrameworkID 的保存器. + * + * @author gaohongtao + */ +@RequiredArgsConstructor +public class FrameworkIDService { + + private final CoordinatorRegistryCenter regCenter; + + /** + * 获取FrameworkID,返回值是一个可选的结果. + * + * @return 获取FrameworkID的可选结果 + */ + public Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.ha; + +/** + * 保存HA相关数据节点. + * + * @author gaohongtao + */ +public class HANode { + + /** + * HA根节点. + */ + public static final String ROOT = "/ha"; + + /** + * FrameworkID保存的节点. + */ + public static final String FRAMEWORK_ID_NODE = ROOT + "/framework_id"; + + /** + * 选举节点. + */ + public static final String ELECTION_NODE = ROOT + "/election"; +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/ha/SchedulerElectionCandidate.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/ha/SchedulerElectionCandidate.java new file mode 100644 index 0000000000..472124f347 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/ha/SchedulerElectionCandidate.java @@ -0,0 +1,49 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.ha; + +import com.dangdang.ddframe.job.cloud.scheduler.mesos.SchedulerService; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.reg.base.ElectionCandidate; + +/** + * 调度器选举候选人. + * + * @author caohao + */ +public class SchedulerElectionCandidate implements ElectionCandidate { + + private final CoordinatorRegistryCenter regCenter; + + private SchedulerService schedulerService; + + public SchedulerElectionCandidate(final CoordinatorRegistryCenter regCenter) { + this.regCenter = regCenter; + } + + @Override + public void startLeadership() { + schedulerService = new SchedulerService(regCenter); + schedulerService.start(); + } + + @Override + public void stopLeadership() { + schedulerService.stop(); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java new file mode 100644 index 0000000000..67c5db7b88 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/mesos/FacadeService.java @@ -0,0 +1,295 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.context.JobContext; +import com.dangdang.ddframe.job.cloud.scheduler.state.failover.FailoverService; +import com.dangdang.ddframe.job.cloud.scheduler.state.failover.FailoverTaskInfo; +import com.dangdang.ddframe.job.cloud.scheduler.state.ready.ReadyService; +import com.dangdang.ddframe.job.cloud.scheduler.state.running.RunningService; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.context.TaskContext.MetaInfo; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Function; +import com.google.common.base.Optional; +import com.google.common.collect.Lists; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * 为Mesos提供的门面服务. + * + * @author zhangliang + * @author caohao + */ +@Slf4j +public class FacadeService { + + private final CloudAppConfigurationService appConfigService; + + private final CloudJobConfigurationService jobConfigService; + + private final ReadyService readyService; + + private final RunningService runningService; + + private final FailoverService failoverService; + + public FacadeService(final CoordinatorRegistryCenter regCenter) { + appConfigService = new CloudAppConfigurationService(regCenter); + jobConfigService = new CloudJobConfigurationService(regCenter); + readyService = new ReadyService(regCenter); + runningService = new RunningService(regCenter); + failoverService = new FailoverService(regCenter); + } + + /** + * 启动门面服务. + */ + public void start() { + log.info("Elastic Job: Start facade service"); + runningService.start(); + } + + /** + * 获取有资格运行的作业. + * + * @return 作业上下文集合 + */ + public CollectionREADY类型的作业为整体, 任意一片运行都视为作业运行. FAILOVER则仅以当前分片运行为运行依据.
+ * + * @param taskContext 任务运行时上下文 + * @return 作业是否在运行 + */ + public boolean isRunning(final TaskContext taskContext) { + return ExecutionType.FAILOVER != taskContext.getType() && !runningService.getRunningTasks(taskContext.getMetaInfo().getJobName()).isEmpty() + || ExecutionType.FAILOVER == taskContext.getType() && runningService.isTaskRunning(taskContext.getMetaInfo()); + } + + /** + * 添加任务主键和主机名称的映射. + * + * @param taskId 任务主键 + * @param hostname 主机名称 + */ + public void addMapping(final String taskId, final String hostname) { + runningService.addMapping(taskId, hostname); + } + + /** + * 根据任务主键获取主机名称并清除该任务. + * + * @param taskId 任务主键 + * @return 删除任务的主机名称 + */ + public String popMapping(final String taskId) { + return runningService.popMapping(taskId); + } + + /** + * 获取待运行的全部任务. + * + * @return 待运行的全部任务 + */ + public Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.context.TaskContext; +import com.netflix.fenzo.ConstraintEvaluator; +import com.netflix.fenzo.TaskRequest; +import com.netflix.fenzo.VMTaskFitnessCalculator; +import lombok.RequiredArgsConstructor; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * 作业任务请求对象. + * + * @author zhangliang + */ +@RequiredArgsConstructor +public final class JobTaskRequest implements TaskRequest { + + private final TaskContext taskContext; + + private final CloudJobConfiguration jobConfig; + + @Override + public String getId() { + return taskContext.getId(); + } + + @Override + public String taskGroupName() { + return ""; + } + + @Override + public double getCPUs() { + return jobConfig.getCpuCount(); + } + + @Override + public double getMemory() { + return jobConfig.getMemoryMB(); + } + + @Override + public double getNetworkMbps() { + return 0; + } + + @Override + public double getDisk() { + return 10d; + } + + @Override + public int getPorts() { + return 1; + } + + @Override + public Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.context.JobContext; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.context.TaskContext; +import com.netflix.fenzo.TaskAssignmentResult; +import com.netflix.fenzo.TaskRequest; +import com.netflix.fenzo.VMAssignmentResult; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; + +/** + * 分配任务行为包. + * + * @author zhangliang + */ +@Slf4j +public final class LaunchingTasks { + + private final Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.netflix.fenzo.VirtualMachineLease; +import com.netflix.fenzo.plugins.VMLeaseObject; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; +import org.apache.mesos.Protos; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +/** + * 资源预占队列. + * + * @author zhangliang + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public class LeasesQueue { + + private static final LeasesQueue INSTANCE = new LeasesQueue(); + + private final BlockingQueue+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.ha.FrameworkIDService; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Function; +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.common.collect.Collections2; +import com.google.common.collect.Lists; +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import com.sun.jersey.api.client.Client; +import lombok.Builder; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.codehaus.jettison.json.JSONArray; +import org.codehaus.jettison.json.JSONException; +import org.codehaus.jettison.json.JSONObject; + +import java.util.Collection; +import java.util.List; + +/** + * Mesos状态服务. + * + * @author gaohongtao + */ +@Slf4j +public class MesosStateService { + + private static String stateUrl; + + private final String frameworkID; + + public MesosStateService(final CoordinatorRegistryCenter regCenter) { + Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.ha.FrameworkIDService; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.StatisticManager; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.event.JobEventBus; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent.Source; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent.State; +import com.netflix.fenzo.TaskScheduler; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.mesos.Protos; +import org.apache.mesos.Scheduler; +import org.apache.mesos.SchedulerDriver; + +import java.util.List; + +/** + * 作业云引擎. + * + * @author zhangliang + */ +@RequiredArgsConstructor +@Slf4j +public final class SchedulerEngine implements Scheduler { + + private final TaskScheduler taskScheduler; + + private final FacadeService facadeService; + + private final JobEventBus jobEventBus; + + private final FrameworkIDService frameworkIDService; + + private final StatisticManager statisticManager; + + @Override + public void registered(final SchedulerDriver schedulerDriver, final Protos.FrameworkID frameworkID, final Protos.MasterInfo masterInfo) { + log.info("call registered"); + frameworkIDService.save(frameworkID.getValue()); + taskScheduler.expireAllLeases(); + MesosStateService.register(masterInfo.getHostname(), masterInfo.getPort()); + } + + @Override + public void reregistered(final SchedulerDriver schedulerDriver, final Protos.MasterInfo masterInfo) { + log.info("call reregistered"); + taskScheduler.expireAllLeases(); + MesosStateService.register(masterInfo.getHostname(), masterInfo.getPort()); + } + + @Override + public void resourceOffers(final SchedulerDriver schedulerDriver, final List+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationListener; +import com.dangdang.ddframe.job.cloud.scheduler.env.BootstrapEnvironment; +import com.dangdang.ddframe.job.cloud.scheduler.env.MesosConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.ha.FrameworkIDService; +import com.dangdang.ddframe.job.cloud.scheduler.producer.ProducerManager; +import com.dangdang.ddframe.job.cloud.scheduler.restful.RestfulService; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.StatisticManager; +import com.dangdang.ddframe.job.event.JobEventBus; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Optional; +import com.google.common.util.concurrent.Service; +import com.netflix.fenzo.TaskScheduler; +import com.netflix.fenzo.VirtualMachineLease; +import com.netflix.fenzo.functions.Action1; +import lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.mesos.MesosSchedulerDriver; +import org.apache.mesos.Protos; +import org.apache.mesos.SchedulerDriver; + +import static com.dangdang.ddframe.job.cloud.scheduler.env.MesosConfiguration.FRAMEWORK_FAILOVER_TIMEOUT; +import static com.dangdang.ddframe.job.cloud.scheduler.env.MesosConfiguration.FRAMEWORK_NAME; + +/** + * 调度服务. + * + * @author zhangliang + * @author caohao + */ +@Slf4j +@AllArgsConstructor +public final class SchedulerService { + + private static final String WEB_UI_PROTOCOL = "http://"; + + private final BootstrapEnvironment env; + + private final FacadeService facadeService; + + private final SchedulerDriver schedulerDriver; + + private final ProducerManager producerManager; + + private final StatisticManager statisticManager; + + private final CloudJobConfigurationListener cloudJobConfigurationListener; + + private final Service taskLaunchScheduledService; + + private final RestfulService restfulService; + + public SchedulerService(final CoordinatorRegistryCenter regCenter) { + env = BootstrapEnvironment.getInstance(); + facadeService = new FacadeService(regCenter); + statisticManager = StatisticManager.getInstance(regCenter, env.getJobEventRdbConfiguration()); + TaskScheduler taskScheduler = getTaskScheduler(); + JobEventBus jobEventBus = getJobEventBus(); + schedulerDriver = getSchedulerDriver(taskScheduler, jobEventBus, new FrameworkIDService(regCenter)); + producerManager = new ProducerManager(schedulerDriver, regCenter); + cloudJobConfigurationListener = new CloudJobConfigurationListener(regCenter, producerManager); + taskLaunchScheduledService = new TaskLaunchScheduledService(schedulerDriver, taskScheduler, facadeService, jobEventBus); + restfulService = new RestfulService(regCenter, env.getRestfulServerConfiguration(), producerManager); + } + + private SchedulerDriver getSchedulerDriver(final TaskScheduler taskScheduler, final JobEventBus jobEventBus, final FrameworkIDService frameworkIDService) { + MesosConfiguration mesosConfig = env.getMesosConfiguration(); + Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import lombok.NoArgsConstructor; + +import java.util.HashSet; +import java.util.Set; + +/** + * Mesos所支持的压缩类型. + * + * @author zhangliang + */ +@NoArgsConstructor +public final class SupportedExtractionType { + + private static final Set+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration; +import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import com.dangdang.ddframe.job.executor.handler.JobProperties; +import lombok.RequiredArgsConstructor; +import org.apache.commons.lang3.SerializationUtils; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * 随任务传递的数据. + * + * @author zhangliang + */ +@RequiredArgsConstructor +public final class TaskInfoData { + + private final ShardingContexts shardingContexts; + + private final CloudJobConfiguration jobConfig; + + /** + * 序列化. + * + * @return 序列化后的字节数组 + */ + public byte[] serialize() { + LinkedHashMap+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.mesos; + +import com.dangdang.ddframe.job.api.JobType; +import com.dangdang.ddframe.job.cloud.scheduler.env.BootstrapEnvironment; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfiguration; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.event.JobEventBus; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent.Source; +import com.dangdang.ddframe.job.executor.ShardingContexts; +import com.dangdang.ddframe.job.util.config.ShardingItemParameters; +import com.dangdang.ddframe.job.util.json.GsonFactory; +import com.google.common.base.Joiner; +import com.google.common.base.Optional; +import com.google.common.util.concurrent.AbstractScheduledService; +import com.google.protobuf.ByteString; +import com.netflix.fenzo.TaskAssignmentResult; +import com.netflix.fenzo.TaskScheduler; +import com.netflix.fenzo.VMAssignmentResult; +import com.netflix.fenzo.VirtualMachineLease; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.exec.CommandLine; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.mesos.Protos; +import org.apache.mesos.Protos.OfferID; +import org.apache.mesos.Protos.TaskInfo; +import org.apache.mesos.SchedulerDriver; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.TimeUnit; + +/** + * 任务提交调度服务. + * + * @author zhangliang + * @author gaohongtao + */ +@RequiredArgsConstructor +@Slf4j +public class TaskLaunchScheduledService extends AbstractScheduledService { + + private final SchedulerDriver schedulerDriver; + + private final TaskScheduler taskScheduler; + + private final FacadeService facadeService; + + private final JobEventBus jobEventBus; + + private final BootstrapEnvironment env = BootstrapEnvironment.getInstance(); + + @Override + protected String serviceName() { + return "task-launch-processor"; + } + + @Override + protected Scheduler scheduler() { + return Scheduler.newFixedDelaySchedule(2, 10, TimeUnit.SECONDS); + } + + @Override + protected void startUp() throws Exception { + log.info("Elastic Job: Start {}", serviceName()); + } + + @Override + protected void shutDown() throws Exception { + log.info("Elastic Job: Stop {}", serviceName()); + } + + @Override + protected void runOneIteration() throws Exception { + try { + LaunchingTasks launchingTasks = new LaunchingTasks(facadeService.getEligibleJobContext()); + List+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.producer; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.state.ready.ReadyService; +import com.dangdang.ddframe.job.cloud.scheduler.state.running.RunningService; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.exception.AppConfigurationException; +import com.dangdang.ddframe.job.exception.JobConfigurationException; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Function; +import com.google.common.base.Optional; +import com.google.common.base.Predicate; +import com.google.common.collect.Collections2; +import com.google.common.collect.Iterators; +import com.google.common.collect.Lists; +import lombok.extern.slf4j.Slf4j; +import org.apache.mesos.Protos; +import org.apache.mesos.SchedulerDriver; + +import java.util.Collection; +import java.util.Collections; +import java.util.Set; + +/** + * 发布任务作业调度管理器. + * + * @author caohao + * @author zhangliang + */ +@Slf4j +public class ProducerManager { + + private final CloudAppConfigurationService appConfigService; + + private final CloudJobConfigurationService configService; + + private final ReadyService readyService; + + private final RunningService runningService; + + private final TransientProducerScheduler transientProducerScheduler; + + private final SchedulerDriver schedulerDriver; + + public ProducerManager(final SchedulerDriver schedulerDriver, final CoordinatorRegistryCenter regCenter) { + this.schedulerDriver = schedulerDriver; + appConfigService = new CloudAppConfigurationService(regCenter); + configService = new CloudJobConfigurationService(regCenter); + readyService = new ReadyService(regCenter); + runningService = new RunningService(regCenter); + transientProducerScheduler = new TransientProducerScheduler(readyService); + } + + /** + * 启动作业调度器. + */ + public void startup() { + log.info("Start producer manager"); + transientProducerScheduler.start(); + for (CloudJobConfiguration each : configService.loadAll()) { + schedule(each); + } + } + + /** + * 注册作业. + * + * @param jobConfig 作业配置 + */ + public void register(final CloudJobConfiguration jobConfig) { + Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.producer; + +import org.quartz.JobKey; + +import java.util.Collections; +import java.util.List; +import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * 瞬时作业生成器数据访问对象. + * + * @author caohao + * @author zhangliang + */ +class TransientProducerRepository { + + private final ConcurrentHashMap+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.producer; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.state.ready.ReadyService; +import com.dangdang.ddframe.job.exception.JobSystemException; +import lombok.Setter; +import org.quartz.CronScheduleBuilder; +import org.quartz.Job; +import org.quartz.JobBuilder; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.quartz.JobKey; +import org.quartz.Scheduler; +import org.quartz.SchedulerException; +import org.quartz.Trigger; +import org.quartz.TriggerBuilder; +import org.quartz.TriggerKey; +import org.quartz.impl.StdSchedulerFactory; +import org.quartz.plugins.management.ShutdownHookPlugin; +import org.quartz.simpl.SimpleThreadPool; + +import java.util.List; +import java.util.Properties; + +/** + * 发布瞬时作业任务的调度器. + * + * @author caohao + */ +class TransientProducerScheduler { + + private final TransientProducerRepository repository; + + private final ReadyService readyService; + + private Scheduler scheduler; + + TransientProducerScheduler(final ReadyService readyService) { + repository = new TransientProducerRepository(); + this.readyService = readyService; + } + + void start() { + scheduler = getScheduler(); + try { + scheduler.start(); + } catch (final SchedulerException ex) { + throw new JobSystemException(ex); + } + } + + private Scheduler getScheduler() { + StdSchedulerFactory factory = new StdSchedulerFactory(); + try { + factory.initialize(getQuartzProperties()); + return factory.getScheduler(); + } catch (final SchedulerException ex) { + throw new JobSystemException(ex); + } + } + + private Properties getQuartzProperties() { + Properties result = new Properties(); + result.put("org.quartz.threadPool.class", SimpleThreadPool.class.getName()); + result.put("org.quartz.threadPool.threadCount", Integer.toString(Runtime.getRuntime().availableProcessors() * 2)); + result.put("org.quartz.scheduler.instanceName", "ELASTIC_JOB_CLOUD_TRANSIENT_PRODUCER"); + result.put("org.quartz.plugin.shutdownhook.class", ShutdownHookPlugin.class.getName()); + result.put("org.quartz.plugin.shutdownhook.cleanShutdown", Boolean.TRUE.toString()); + return result; + } + + // TODO 并发优化 + synchronized void register(final CloudJobConfiguration jobConfig) { + String cron = jobConfig.getTypeConfig().getCoreConfig().getCron(); + JobKey jobKey = buildJobKey(cron); + repository.put(jobKey, jobConfig.getJobName()); + try { + if (!scheduler.checkExists(jobKey)) { + scheduler.scheduleJob(buildJobDetail(jobKey), buildTrigger(jobKey.getName())); + } + } catch (final SchedulerException ex) { + throw new JobSystemException(ex); + } + } + + private JobDetail buildJobDetail(final JobKey jobKey) { + JobDetail result = JobBuilder.newJob(ProducerJob.class).withIdentity(jobKey).build(); + result.getJobDataMap().put("repository", repository); + result.getJobDataMap().put("readyService", readyService); + return result; + } + + private Trigger buildTrigger(final String cron) { + return TriggerBuilder.newTrigger().withIdentity(cron).withSchedule(CronScheduleBuilder.cronSchedule(cron).withMisfireHandlingInstructionDoNothing()).build(); + } + + void deregister(final CloudJobConfiguration jobConfig) { + repository.remove(jobConfig.getJobName()); + String cron = jobConfig.getTypeConfig().getCoreConfig().getCron(); + if (!repository.containsKey(buildJobKey(cron))) { + try { + scheduler.unscheduleJob(TriggerKey.triggerKey(cron)); + } catch (final SchedulerException ex) { + throw new JobSystemException(ex); + } + } + } + + private JobKey buildJobKey(final String cron) { + return JobKey.jobKey(cron); + } + + void shutdown() { + try { + if (null != scheduler && !scheduler.isShutdown()) { + scheduler.shutdown(); + } + } catch (final SchedulerException ex) { + throw new JobSystemException(ex); + } + repository.removeAll(); + } + + @Setter + public static final class ProducerJob implements Job { + + private TransientProducerRepository repository; + + private ReadyService readyService; + + @Override + public void execute(final JobExecutionContext context) throws JobExecutionException { + List+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.restful; + +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfigurationGsonFactory; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfigurationService; +import com.dangdang.ddframe.job.exception.AppConfigurationException; +import com.dangdang.ddframe.job.exception.JobSystemException; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.util.json.GsonFactory; +import com.google.common.base.Optional; + +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.core.MediaType; +import java.util.Collection; + +/** + * 云作业App的REST API. + * + * @author caohao + */ +@Path("/app") +public final class CloudAppRestfulApi { + + private static CoordinatorRegistryCenter regCenter; + + private final CloudAppConfigurationService configService; + + public CloudAppRestfulApi() { + configService = new CloudAppConfigurationService(regCenter); + } + + /** + * 初始化. + * + * @param regCenter 注册中心 + */ + public static void init(final CoordinatorRegistryCenter regCenter) { + CloudAppRestfulApi.regCenter = regCenter; + GsonFactory.registerTypeAdapter(CloudAppConfiguration.class, new CloudAppConfigurationGsonFactory.CloudAppConfigurationGsonTypeAdapter()); + } + + /** + * 注册云作业APP配置. + * + * @param appConfig 云作业APP配置 + */ + @POST + @Consumes(MediaType.APPLICATION_JSON) + public void register(final CloudAppConfiguration appConfig) { + Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.restful; + +import com.dangdang.ddframe.job.cloud.scheduler.env.BootstrapEnvironment; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationGsonFactory; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.cloud.scheduler.mesos.FacadeService; +import com.dangdang.ddframe.job.cloud.scheduler.producer.ProducerManager; +import com.dangdang.ddframe.job.cloud.scheduler.state.failover.FailoverTaskInfo; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.StatisticManager; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbSearch; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbSearch.Condition; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbSearch.Result; +import com.dangdang.ddframe.job.event.type.JobExecutionEvent; +import com.dangdang.ddframe.job.event.type.JobStatusTraceEvent; +import com.dangdang.ddframe.job.exception.JobConfigurationException; +import com.dangdang.ddframe.job.exception.JobSystemException; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.statistics.StatisticInterval; +import com.dangdang.ddframe.job.statistics.type.job.JobExecutionTypeStatistics; +import com.dangdang.ddframe.job.statistics.type.job.JobRegisterStatistics; +import com.dangdang.ddframe.job.statistics.type.job.JobRunningStatistics; +import com.dangdang.ddframe.job.statistics.type.job.JobTypeStatistics; +import com.dangdang.ddframe.job.statistics.type.task.TaskResultStatistics; +import com.dangdang.ddframe.job.statistics.type.task.TaskRunningStatistics; +import com.dangdang.ddframe.job.util.json.GsonFactory; +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import lombok.extern.slf4j.Slf4j; + +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.UriInfo; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; + +/** + * 作业云Job的REST API. + * + * @author zhangliang + * @author liguangyun + */ +@Path("/job") +@Slf4j +public final class CloudJobRestfulApi { + + private static CoordinatorRegistryCenter regCenter; + + private static JobEventRdbSearch jobEventRdbSearch; + + private static ProducerManager producerManager; + + private final CloudJobConfigurationService configService; + + private final FacadeService facadeService; + + private final StatisticManager statisticManager; + + public CloudJobRestfulApi() { + Preconditions.checkNotNull(regCenter); + configService = new CloudJobConfigurationService(regCenter); + facadeService = new FacadeService(regCenter); + Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.restful; + +import com.dangdang.ddframe.job.cloud.scheduler.mesos.MesosStateService; +import com.dangdang.ddframe.job.cloud.scheduler.producer.ProducerManager; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.gson.JsonArray; +import lombok.extern.slf4j.Slf4j; +import org.codehaus.jettison.json.JSONException; + +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.QueryParam; + +/** + * 作业云维护服务. + * + * @author gaohongtao. + */ +@Path("/operate") +@Slf4j +public final class CloudOperationRestfulApi { + + private static ProducerManager producerManager; + + private static MesosStateService mesosStateService; + + private static final long RECONCILE_INTERVAL = 10 * 1000; + + private static long lastReconcileTime; + + /** + * 初始化. + * + * @param producerManager 生产管理器 + */ + public static void init(final CoordinatorRegistryCenter regCenter, final ProducerManager producerManager) { + CloudOperationRestfulApi.producerManager = producerManager; + CloudOperationRestfulApi.mesosStateService = new MesosStateService(regCenter); + } + + /** + * 显示协调服务. + * + * @param taskId 可选参数,如果存在taskId那么只针对该task进行协调.如果不传入taskId,对所有的服务进行协调 + */ + @POST + @Path("/reconcile/explicit") + public void explicitReconcile(@QueryParam("taskId") final String taskId) { + synchronized (CloudOperationRestfulApi.class) { + validReconcileInterval(); + if (Strings.isNullOrEmpty(taskId)) { + producerManager.explicitReconcile(); + } else { + producerManager.explicitReconcile(taskId); + } + } + } + + /** + * 隐式协调服务. + */ + @POST + @Path("/reconcile/implicit") + public void implicitReconcile() { + synchronized (CloudOperationRestfulApi.class) { + validReconcileInterval(); + producerManager.implicitReconcile(); + } + } + + private void validReconcileInterval() { + if (System.currentTimeMillis() < lastReconcileTime + RECONCILE_INTERVAL) { + throw new RuntimeException("Repeat explicitReconcile"); + } + lastReconcileTime = System.currentTimeMillis(); + } + + /** + * 获取作业云App的沙箱信息. + * + * @param appName 云作业App配置名称 + * @return 沙箱信息 + */ + @GET + @Path("/sandbox") + public JsonArray sandbox(@QueryParam("appName") final String appName) throws JSONException { + Preconditions.checkArgument(!Strings.isNullOrEmpty(appName), "Lack param 'appName'"); + return mesosStateService.sandbox(appName); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/RestfulService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/RestfulService.java new file mode 100644 index 0000000000..ca6ef520a4 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/restful/RestfulService.java @@ -0,0 +1,70 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.restful; + +import com.dangdang.ddframe.job.cloud.scheduler.env.RestfulServerConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationGsonFactory; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.app.CloudAppConfigurationGsonFactory; +import com.dangdang.ddframe.job.cloud.scheduler.producer.ProducerManager; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.restful.RestfulServer; +import com.dangdang.ddframe.job.util.json.GsonFactory; +import com.google.common.base.Optional; + +/** + * 云作业Restful服务. + * + * @author caohao + */ +public class RestfulService { + + private static final String CONSOLE_PATH = "console"; + + private final RestfulServer restfulServer; + + public RestfulService(final CoordinatorRegistryCenter regCenter, final RestfulServerConfiguration config, final ProducerManager producerManager) { + restfulServer = new RestfulServer(config.getPort()); + GsonFactory.registerTypeAdapter(CloudJobConfiguration.class, new CloudJobConfigurationGsonFactory.CloudJobConfigurationGsonTypeAdapter()); + GsonFactory.registerTypeAdapter(CloudAppConfiguration.class, new CloudAppConfigurationGsonFactory.CloudAppConfigurationGsonTypeAdapter()); + CloudJobRestfulApi.init(regCenter, producerManager); + CloudAppRestfulApi.init(regCenter); + CloudOperationRestfulApi.init(regCenter, producerManager); + } + + /** + * 启动Restful服务. + */ + public void start() { + try { + restfulServer.start(RestfulService.class.getPackage().getName(), Optional.of(CONSOLE_PATH)); + //CHECKSTYLE:OFF + } catch (final Exception ex) { + //CHECKSTYLE:ON + throw new RuntimeException(ex.getCause()); + } + } + + /** + * 停止Restful服务. + */ + public void stop() { + restfulServer.stop(); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/StateNode.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/StateNode.java new file mode 100644 index 0000000000..38691f3a1f --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/StateNode.java @@ -0,0 +1,35 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +/** + * 状态根节点路径. + * + * @author zhangliang + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class StateNode { + + /** + * 状态根节点. + */ + public static final String ROOT = "/state"; +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/failover/FailoverNode.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/failover/FailoverNode.java new file mode 100644 index 0000000000..72d0519e04 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/failover/FailoverNode.java @@ -0,0 +1,46 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state.failover; + +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.cloud.scheduler.state.StateNode; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +/** + * 待失效转移任务队列节点路径. + * + * @author zhangliang + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +final class FailoverNode { + + static final String ROOT = StateNode.ROOT + "/failover"; + + private static final String FAILOVER_JOB = ROOT + "/%s"; + + private static final String FAILOVER_TASK = FAILOVER_JOB + "/%s"; + + static String getFailoverJobNodePath(final String jobName) { + return String.format(FAILOVER_JOB, jobName); + } + + static String getFailoverTaskNodePath(final String taskMetaInfo) { + return String.format(FAILOVER_TASK, TaskContext.MetaInfo.from(taskMetaInfo).getJobName(), taskMetaInfo); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/failover/FailoverService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/failover/FailoverService.java new file mode 100644 index 0000000000..2e66cde865 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/failover/FailoverService.java @@ -0,0 +1,191 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state.failover; + +import com.dangdang.ddframe.job.cloud.scheduler.env.BootstrapEnvironment; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.context.JobContext; +import com.dangdang.ddframe.job.cloud.scheduler.state.running.RunningService; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.context.TaskContext.MetaInfo; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Charsets; +import com.google.common.base.Optional; +import com.google.common.base.Strings; +import com.google.common.hash.HashCode; +import com.google.common.hash.Hashing; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * 失效转移队列服务. + * + * @author zhangliang + */ +@Slf4j +public class FailoverService { + + private final BootstrapEnvironment env = BootstrapEnvironment.getInstance(); + + private final CoordinatorRegistryCenter regCenter; + + private final CloudJobConfigurationService configService; + + private final RunningService runningService; + + public FailoverService(final CoordinatorRegistryCenter regCenter) { + this.regCenter = regCenter; + configService = new CloudJobConfigurationService(regCenter); + runningService = new RunningService(regCenter); + } + + /** + * 将任务放入失效转移队列. + * + * @param taskContext 任务运行时上下文 + */ + public void add(final TaskContext taskContext) { + if (regCenter.getNumChildren(FailoverNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { + log.warn("Cannot add job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); + return; + } + String failoverTaskNodePath = FailoverNode.getFailoverTaskNodePath(taskContext.getMetaInfo().toString()); + if (!regCenter.isExisted(failoverTaskNodePath) && !runningService.isTaskRunning(taskContext.getMetaInfo())) { + // TODO Daemon类型作业增加存储是否立即失效转移 + regCenter.persist(failoverTaskNodePath, taskContext.getId()); + } + } + + /** + * 从失效转移队列中获取所有有资格执行的作业上下文. + * + * @return 有资格执行的作业上下文集合 + */ + public Collection+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state.failover; + +import com.dangdang.ddframe.job.context.TaskContext.MetaInfo; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * 待失效转移任务节点信息. + * + * @author liguangyun + */ +@RequiredArgsConstructor +@Getter +public final class FailoverTaskInfo { + + private final MetaInfo taskInfo; + + private final String originalTaskId; +} diff --git a/elastic-job-console/src/main/java/com/dangdang/ddframe/job/console/domain/RegistryCenterClient.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/ready/ReadyNode.java similarity index 53% rename from elastic-job-console/src/main/java/com/dangdang/ddframe/job/console/domain/RegistryCenterClient.java rename to elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/ready/ReadyNode.java index dd178edc66..bf93777cc6 100644 --- a/elastic-job-console/src/main/java/com/dangdang/ddframe/job/console/domain/RegistryCenterClient.java +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/ready/ReadyNode.java @@ -1,12 +1,12 @@ -/** +/* * Copyright 1999-2015 dangdang.com. ** Licensed 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 - * + * + * 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. @@ -15,30 +15,25 @@ *
*/ -package com.dangdang.ddframe.job.console.domain; - -import java.io.Serializable; +package com.dangdang.ddframe.job.cloud.scheduler.state.ready; -import lombok.Getter; +import com.dangdang.ddframe.job.cloud.scheduler.state.StateNode; +import lombok.AccessLevel; import lombok.NoArgsConstructor; -import lombok.Setter; -import org.apache.curator.framework.CuratorFramework; - -@Getter -@Setter -@NoArgsConstructor -public final class RegistryCenterClient implements Serializable { - - private static final long serialVersionUID = -946258964014121184L; - - private String name; +/** + * 待运行作业队列节点路径. + * + * @author zhangliang + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +final class ReadyNode { - private CuratorFramework curatorClient; + static final String ROOT = StateNode.ROOT + "/ready"; - private boolean connected; + private static final String READY_JOB = ROOT + "/%s"; - public RegistryCenterClient(final String name) { - this.name = name; + static String getReadyJobNodePath(final String jobName) { + return String.format(READY_JOB, jobName); } } diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/ready/ReadyService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/ready/ReadyService.java new file mode 100644 index 0000000000..7d93c12007 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/ready/ReadyService.java @@ -0,0 +1,188 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state.ready; + +import com.dangdang.ddframe.job.cloud.scheduler.env.BootstrapEnvironment; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.context.ExecutionType; +import com.dangdang.ddframe.job.cloud.scheduler.context.JobContext; +import com.dangdang.ddframe.job.cloud.scheduler.state.running.RunningService; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Function; +import com.google.common.base.Optional; +import com.google.common.base.Strings; +import com.google.common.collect.Collections2; +import lombok.extern.slf4j.Slf4j; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 待运行作业队列服务. + * + * @author zhangliang + * @author liguangyun + */ +@Slf4j +public class ReadyService { + + private final BootstrapEnvironment env = BootstrapEnvironment.getInstance(); + + private final CoordinatorRegistryCenter regCenter; + + private final CloudJobConfigurationService configService; + + private final RunningService runningService; + + public ReadyService(final CoordinatorRegistryCenter regCenter) { + this.regCenter = regCenter; + configService = new CloudJobConfigurationService(regCenter); + runningService = new RunningService(regCenter); + } + + /** + * 将瞬时作业放入待执行队列. + * + * @param jobName 作业名称 + */ + public void addTransient(final String jobName) { + if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { + log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); + return; + } + Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state.running; + +import com.dangdang.ddframe.job.cloud.scheduler.state.StateNode; +import com.dangdang.ddframe.job.context.TaskContext; +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +/** + * 运行中任务节点路径. + * + * @author zhangliang + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +final class RunningNode { + + static final String ROOT = StateNode.ROOT + "/running"; + + private static final String RUNNING_JOB = ROOT + "/%s"; + + private static final String RUNNING_TASK = RUNNING_JOB + "/%s"; + + static String getRunningJobNodePath(final String jobName) { + return String.format(RUNNING_JOB, jobName); + } + + static String getRunningTaskNodePath(final String taskMetaInfo) { + return String.format(RUNNING_TASK, TaskContext.MetaInfo.from(taskMetaInfo).getJobName(), taskMetaInfo); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/running/RunningService.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/running/RunningService.java new file mode 100644 index 0000000000..963e6fd5ae --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/state/running/RunningService.java @@ -0,0 +1,268 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.state.running; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.google.common.base.Function; +import com.google.common.base.Optional; +import com.google.common.base.Predicate; +import com.google.common.collect.Iterators; +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArraySet; + +/** + * 任务运行时服务. + * + * @author zhangliang + */ +@RequiredArgsConstructor +public class RunningService { + + private static final int TASK_INITIAL_SIZE = 1024; + + // TODO 使用JMX导出 + @Getter + private static final ConcurrentHashMap+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics; + +import com.dangdang.ddframe.job.api.JobType; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfiguration; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobExecutionType; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.job.JobRunningStatisticJob; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.job.RegisteredJobStatisticJob; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.job.TaskResultStatisticJob; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.util.StatisticTimeUtils; +import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration; +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.statistics.StatisticInterval; +import com.dangdang.ddframe.job.statistics.rdb.StatisticRdbRepository; +import com.dangdang.ddframe.job.statistics.type.job.JobExecutionTypeStatistics; +import com.dangdang.ddframe.job.statistics.type.job.JobRegisterStatistics; +import com.dangdang.ddframe.job.statistics.type.job.JobRunningStatistics; +import com.dangdang.ddframe.job.statistics.type.job.JobTypeStatistics; +import com.dangdang.ddframe.job.statistics.type.task.TaskResultStatistics; +import com.dangdang.ddframe.job.statistics.type.task.TaskRunningStatistics; +import com.google.common.base.Optional; +import lombok.AccessLevel; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import java.sql.SQLException; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 统计作业调度管理器. + * + * @author liguangyun + */ +@Slf4j +@RequiredArgsConstructor(access = AccessLevel.PRIVATE) +public class StatisticManager { + + private static volatile StatisticManager instance; + + private final CoordinatorRegistryCenter registryCenter; + + private final CloudJobConfigurationService configurationService; + + private final Optional+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics; + +import java.util.Map; +import java.util.Properties; + +import org.quartz.JobDetail; +import org.quartz.Scheduler; +import org.quartz.SchedulerException; +import org.quartz.impl.StdSchedulerFactory; +import org.quartz.plugins.management.ShutdownHookPlugin; +import org.quartz.simpl.SimpleThreadPool; + +import com.dangdang.ddframe.job.cloud.scheduler.statistics.job.StatisticJob; +import com.dangdang.ddframe.job.exception.JobStatisticException; + +/** + * 统计作业调度器. + * + * @author liguangyun + */ +class StatisticsScheduler { + + private final StdSchedulerFactory factory; + + private Scheduler scheduler; + + /** + * 构造函数. + */ + StatisticsScheduler() { + factory = new StdSchedulerFactory(); + try { + factory.initialize(getQuartzProperties()); + } catch (final SchedulerException ex) { + throw new JobStatisticException(ex); + } + } + + private Properties getQuartzProperties() { + Properties result = new Properties(); + result.put("org.quartz.threadPool.class", SimpleThreadPool.class.getName()); + result.put("org.quartz.threadPool.threadCount", Integer.toString(1)); + result.put("org.quartz.scheduler.instanceName", "ELASTIC_JOB_CLOUD_STATISTICS_SCHEDULER"); + result.put("org.quartz.plugin.shutdownhook.class", ShutdownHookPlugin.class.getName()); + result.put("org.quartz.plugin.shutdownhook.cleanShutdown", Boolean.TRUE.toString()); + return result; + } + + /** + * 启动调度器. + */ + void start() { + try { + scheduler = factory.getScheduler(); + scheduler.start(); + } catch (final SchedulerException ex) { + throw new JobStatisticException(ex); + } + } + + /** + * 注册统计作业. + * + * @param statisticJob 统计作业 + */ + void register(final StatisticJob statisticJob) { + try { + JobDetail jobDetail = statisticJob.buildJobDetail(); + Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics; + +import java.util.concurrent.atomic.AtomicInteger; + +/** + * 统计元数据. + * + * @author liguangyun + */ +public class TaskResultMetaData { + + private final AtomicInteger successCount; + + private final AtomicInteger failedCount; + + /** + * 构造函数. + */ + public TaskResultMetaData() { + successCount = new AtomicInteger(0); + failedCount = new AtomicInteger(0); + } + + /** + * 增加并获取成功数. + * + * @return 成功数 + */ + public int incrementAndGetSuccessCount() { + return successCount.incrementAndGet(); + } + + /** + * 增加并获取失败数. + * + * @return 失败数 + */ + public int incrementAndGetFailedCount() { + return failedCount.incrementAndGet(); + } + + /** + * 获取成功数. + * + * @return 成功数 + */ + public int getSuccessCount() { + return successCount.get(); + } + + /** + * 获取失败数. + * + * @return 失败数 + */ + public int getFailedCount() { + return failedCount.get(); + } + + /** + * 重置成功数、失败数. + */ + public void reset() { + successCount.set(0); + failedCount.set(0); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/statistics/job/AbstractStatisticJob.java b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/statistics/job/AbstractStatisticJob.java new file mode 100644 index 0000000000..85ec8fff52 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/java/com/dangdang/ddframe/job/cloud/scheduler/statistics/job/AbstractStatisticJob.java @@ -0,0 +1,54 @@ +/* + * Copyright 1999-2015 dangdang.com. + *+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics.job; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.List; + +import com.dangdang.ddframe.job.cloud.scheduler.statistics.util.StatisticTimeUtils; +import com.dangdang.ddframe.job.statistics.StatisticInterval; + +/** + * 统计作业抽象类. + * + * @author liguangyun + */ +abstract class AbstractStatisticJob implements StatisticJob { + + String getJobName() { + return this.getClass().getSimpleName(); + } + + String getTriggerName() { + return this.getClass().getSimpleName() + "Trigger"; + } + + List+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics.job; + +import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter; +import com.dangdang.ddframe.job.cloud.scheduler.state.running.RunningService; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.util.StatisticTimeUtils; +import com.dangdang.ddframe.job.context.TaskContext; +import com.dangdang.ddframe.job.statistics.StatisticInterval; +import com.dangdang.ddframe.job.statistics.rdb.StatisticRdbRepository; +import com.dangdang.ddframe.job.statistics.type.job.JobRunningStatistics; +import com.dangdang.ddframe.job.statistics.type.task.TaskRunningStatistics; +import com.google.common.base.Optional; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; +import org.quartz.CronScheduleBuilder; +import org.quartz.JobBuilder; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.quartz.Trigger; +import org.quartz.TriggerBuilder; + +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * 运行中的任务统计作业. + * + * @author liguangyun + */ +@Setter +@NoArgsConstructor +@Slf4j +public class JobRunningStatisticJob extends AbstractStatisticJob { + + private RunningService runningService; + + private StatisticRdbRepository repository; + + private final StatisticInterval execInterval = StatisticInterval.MINUTE; + + /** + * 构造函数. + * @param registryCenter 注册中心 + * @param rdbRepository 基于rdb的数据仓库对象 + */ + public JobRunningStatisticJob(final CoordinatorRegistryCenter registryCenter, final StatisticRdbRepository rdbRepository) { + runningService = new RunningService(registryCenter); + this.repository = rdbRepository; + } + + @Override + public JobDetail buildJobDetail() { + return JobBuilder.newJob(this.getClass()).withIdentity(getJobName()).build(); + } + + @Override + public Trigger buildTrigger() { + return TriggerBuilder.newTrigger() + .withIdentity(getTriggerName()) + .withSchedule(CronScheduleBuilder.cronSchedule(execInterval.getCron()) + .withMisfireHandlingInstructionDoNothing()).build(); + } + + @Override + public Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics.job; + +import com.dangdang.ddframe.job.cloud.scheduler.config.job.CloudJobConfigurationService; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.util.StatisticTimeUtils; +import com.dangdang.ddframe.job.statistics.StatisticInterval; +import com.dangdang.ddframe.job.statistics.rdb.StatisticRdbRepository; +import com.dangdang.ddframe.job.statistics.type.job.JobRegisterStatistics; +import com.google.common.base.Optional; +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; +import org.quartz.CronScheduleBuilder; +import org.quartz.JobBuilder; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.quartz.Trigger; +import org.quartz.TriggerBuilder; + +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 已注册作业统计作业. + * + * @author liguangyun + */ +@Setter +@NoArgsConstructor +@AllArgsConstructor +@Slf4j +public class RegisteredJobStatisticJob extends AbstractStatisticJob { + + private CloudJobConfigurationService configurationService; + + private StatisticRdbRepository repository; + + private final StatisticInterval execInterval = StatisticInterval.DAY; + + @Override + public JobDetail buildJobDetail() { + return JobBuilder.newJob(this.getClass()).withIdentity(getJobName()).build(); + } + + @Override + public Trigger buildTrigger() { + return TriggerBuilder.newTrigger() + .withIdentity(getTriggerName()) + .withSchedule(CronScheduleBuilder.cronSchedule(execInterval.getCron()) + .withMisfireHandlingInstructionDoNothing()).build(); + } + + @Override + public Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics.job; + +import java.util.Map; + +import org.quartz.Job; +import org.quartz.JobDetail; +import org.quartz.Trigger; + +/** + * 统计作业. + * + * @author liguangyun + */ +public interface StatisticJob extends Job { + + /** + * 构建JobDetail. + * + * @return JobDetail对象 + */ + JobDetail buildJobDetail(); + + /** + * 构建Trigger. + * + * @return Trigger对象 + */ + Trigger buildTrigger(); + + /** + * 获取对象属性Map. + * + * @return 对象属性Map，KEY为属性名称，VALUE为属性实例 + */ + Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics.job; + +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.quartz.CronScheduleBuilder; +import org.quartz.JobBuilder; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.quartz.Trigger; +import org.quartz.TriggerBuilder; + +import com.dangdang.ddframe.job.cloud.scheduler.statistics.TaskResultMetaData; +import com.dangdang.ddframe.job.cloud.scheduler.statistics.util.StatisticTimeUtils; +import com.dangdang.ddframe.job.statistics.StatisticInterval; +import com.dangdang.ddframe.job.statistics.rdb.StatisticRdbRepository; +import com.dangdang.ddframe.job.statistics.type.task.TaskResultStatistics; +import com.google.common.base.Optional; + +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; + +/** + * 任务运行结果统计作业. + * + * @author liguangyun + */ +@Setter +@NoArgsConstructor +@AllArgsConstructor +@Slf4j +public class TaskResultStatisticJob extends AbstractStatisticJob { + + private StatisticInterval statisticInterval; + + private TaskResultMetaData sharedData; + + private StatisticRdbRepository repository; + + @Override + public JobDetail buildJobDetail() { + JobDetail result = JobBuilder.newJob(this.getClass()).withIdentity(getJobName() + "_" + statisticInterval).build(); + result.getJobDataMap().put("statisticUnit", statisticInterval); + return result; + } + + @Override + public Trigger buildTrigger() { + return TriggerBuilder.newTrigger() + .withIdentity(getTriggerName() + "_" + statisticInterval) + .withSchedule(CronScheduleBuilder.cronSchedule(statisticInterval.getCron()) + .withMisfireHandlingInstructionDoNothing()).build(); + } + + @Override + public Map+ * Licensed 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 com.dangdang.ddframe.job.cloud.scheduler.statistics.util; + +import java.util.Calendar; +import java.util.Date; + +import com.dangdang.ddframe.job.statistics.StatisticInterval; + +import lombok.AccessLevel; +import lombok.NoArgsConstructor; + +/** + * 统计时间工具类. + * + * @author liguangyun + */ +@NoArgsConstructor(access = AccessLevel.PRIVATE) +public final class StatisticTimeUtils { + + /** + * 获取以interval为时间间隔单位的统计时间. + * + * @param interval 时间间隔 + * @return 时间对象 + */ + public static Date getCurrentStatisticTime(final StatisticInterval interval) { + return getStatisticTime(interval, 0); + } + + /** + * 偏移offset个时间间隔单位，获取以interval为时间间隔单位的统计时间. + * offset为负数表示时间向过去偏移，正数表示向未来偏移. + * + * @param interval 时间间隔 + * @param offset 时间偏移量 + * @return 时间对象 + */ + public static Date getStatisticTime(final StatisticInterval interval, final int offset) { + Calendar calendar = Calendar.getInstance(); + calendar.set(Calendar.MILLISECOND, 0); + calendar.set(Calendar.SECOND, 0); + switch (interval) { + case DAY: + calendar.set(Calendar.MINUTE, 0); + calendar.set(Calendar.HOUR_OF_DAY, 0); + calendar.add(Calendar.DATE, offset); + break; + case HOUR: + calendar.set(Calendar.MINUTE, 0); + calendar.add(Calendar.HOUR_OF_DAY, offset); + break; + case MINUTE: + default: + calendar.add(Calendar.MINUTE, offset); + break; + } + return calendar.getTime(); + } +} diff --git a/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/resources/assembly/assembly.xml b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/resources/assembly/assembly.xml new file mode 100644 index 0000000000..30273561a3 --- /dev/null +++ b/elastic-job-cloud/elastic-job-cloud-scheduler/src/main/resources/assembly/assembly.xml @@ -0,0 +1,28 @@ +