IrynaMatveieva
1 year ago
404 changed files with 35555 additions and 4235 deletions
|
Before Width: | Height: | Size: 8.9 KiB After Width: | Height: | Size: 9.0 KiB |
|
Before Width: | Height: | Size: 8.9 KiB After Width: | Height: | Size: 9.0 KiB |
|
Before Width: | Height: | Size: 8.9 KiB After Width: | Height: | Size: 9.0 KiB |
|
Before Width: | Height: | Size: 8.9 KiB After Width: | Height: | Size: 9.0 KiB |
@ -0,0 +1,117 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.controller; |
|||
|
|||
import io.swagger.v3.oas.annotations.Parameter; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.web.bind.annotation.DeleteMapping; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.PostMapping; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.rule.engine.api.JobManager; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.JobId; |
|||
import org.thingsboard.server.common.data.job.Job; |
|||
import org.thingsboard.server.common.data.job.JobFilter; |
|||
import org.thingsboard.server.common.data.job.JobStatus; |
|||
import org.thingsboard.server.common.data.job.JobType; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.controller.ControllerConstants.PAGE_NUMBER_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.PAGE_SIZE_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_DESCRIPTION; |
|||
import static org.thingsboard.server.controller.ControllerConstants.SORT_PROPERTY_DESCRIPTION; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api") |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class JobController extends BaseController { |
|||
|
|||
private final JobManager jobManager; |
|||
|
|||
@GetMapping("/job/{id}") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public Job getJobById(@PathVariable UUID id) throws ThingsboardException { |
|||
JobId jobId = new JobId(id); |
|||
return checkJobId(jobId, Operation.READ); |
|||
} |
|||
|
|||
@GetMapping("/jobs") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public PageData<Job> getJobs(@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) |
|||
@RequestParam int pageSize, |
|||
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) |
|||
@RequestParam int page, |
|||
@Parameter(description = "Case-insensitive 'substring' filter based on job's description") |
|||
@RequestParam(required = false) String textSearch, |
|||
@Parameter(description = SORT_PROPERTY_DESCRIPTION) |
|||
@RequestParam(required = false) String sortProperty, |
|||
@Parameter(description = SORT_ORDER_DESCRIPTION) |
|||
@RequestParam(required = false) String sortOrder, |
|||
@RequestParam(required = false) List<JobType> types, |
|||
@RequestParam(required = false) List<JobStatus> statuses, |
|||
@RequestParam(required = false) List<UUID> entities, |
|||
@RequestParam(required = false) Long startTime, |
|||
@RequestParam(required = false) Long endTime) throws ThingsboardException { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
JobFilter filter = JobFilter.builder() |
|||
.types(types) |
|||
.statuses(statuses) |
|||
.entities(entities) |
|||
.startTime(startTime) |
|||
.endTime(endTime) |
|||
.build(); |
|||
return jobService.findJobsByFilter(getTenantId(), filter, pageLink); |
|||
} |
|||
|
|||
@PostMapping("/job/{id}/cancel") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public void cancelJob(@PathVariable UUID id) throws ThingsboardException { |
|||
JobId jobId = new JobId(id); |
|||
checkJobId(jobId, Operation.WRITE); |
|||
jobManager.cancelJob(getTenantId(), jobId); |
|||
} |
|||
|
|||
@PostMapping("/job/{id}/reprocess") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public void reprocessJob(@PathVariable UUID id) throws ThingsboardException { |
|||
JobId jobId = new JobId(id); |
|||
checkJobId(jobId, Operation.WRITE); |
|||
jobManager.reprocessJob(getTenantId(), jobId); |
|||
} |
|||
|
|||
@DeleteMapping("/job/{id}") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public void deleteJob(@PathVariable UUID id) throws ThingsboardException { |
|||
JobId jobId = new JobId(id); |
|||
checkJobId(jobId, Operation.DELETE); |
|||
jobService.deleteJob(getTenantId(), jobId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.dao.job.JobService; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class JobsDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final JobService jobService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
int deletedCount = jobService.deleteJobsByEntityId(task.getTenantId(), task.getEntityId()); |
|||
log.debug("[{}][{}][{}] Deleted {} jobs", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), deletedCount); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_JOBS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,207 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import jakarta.annotation.PreDestroy; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.ObjectUtils; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardExecutors; |
|||
import org.thingsboard.rule.engine.api.JobManager; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.JobId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.job.Job; |
|||
import org.thingsboard.server.common.data.job.JobResult; |
|||
import org.thingsboard.server.common.data.job.JobStatus; |
|||
import org.thingsboard.server.common.data.job.JobType; |
|||
import org.thingsboard.server.common.data.job.task.Task; |
|||
import org.thingsboard.server.common.data.job.task.TaskResult; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.dao.job.JobService; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TaskProto; |
|||
import org.thingsboard.server.queue.TbQueueCallback; |
|||
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.settings.TasksQueueConfig; |
|||
import org.thingsboard.server.queue.task.JobStatsService; |
|||
import org.thingsboard.server.queue.task.TaskProducerQueueFactory; |
|||
|
|||
import java.util.Arrays; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.function.Function; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Component |
|||
@Slf4j |
|||
public class DefaultJobManager implements JobManager { |
|||
|
|||
private final JobService jobService; |
|||
private final JobStatsService jobStatsService; |
|||
private final PartitionService partitionService; |
|||
private final TasksQueueConfig queueConfig; |
|||
private final Map<JobType, JobProcessor> jobProcessors; |
|||
private final Map<JobType, TbQueueProducer<TbProtoQueueMsg<TaskProto>>> taskProducers; |
|||
private final ExecutorService executor; |
|||
|
|||
public DefaultJobManager(JobService jobService, JobStatsService jobStatsService, PartitionService partitionService, |
|||
TaskProducerQueueFactory queueFactory, TasksQueueConfig queueConfig, |
|||
List<JobProcessor> jobProcessors) { |
|||
this.jobService = jobService; |
|||
this.jobStatsService = jobStatsService; |
|||
this.partitionService = partitionService; |
|||
this.queueConfig = queueConfig; |
|||
this.jobProcessors = jobProcessors.stream().collect(Collectors.toMap(JobProcessor::getType, Function.identity())); |
|||
this.taskProducers = Arrays.stream(JobType.values()).collect(Collectors.toMap(Function.identity(), queueFactory::createTaskProducer)); |
|||
this.executor = ThingsBoardExecutors.newWorkStealingPool(Math.max(4, Runtime.getRuntime().availableProcessors()), getClass()); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Job> submitJob(Job job) { |
|||
log.debug("Submitting job: {}", job); |
|||
return Futures.submit(() -> jobService.saveJob(job.getTenantId(), job), executor); |
|||
} |
|||
|
|||
@Override |
|||
public void onJobUpdate(Job job) { |
|||
JobStatus status = job.getStatus(); |
|||
switch (status) { |
|||
case PENDING -> { |
|||
executor.execute(() -> { |
|||
try { |
|||
processJob(job); |
|||
} catch (Throwable e) { |
|||
log.error("Failed to process job update: {}", job, e); |
|||
} |
|||
}); |
|||
} |
|||
case COMPLETED, FAILED -> { |
|||
executor.execute(() -> { |
|||
try { |
|||
getJobProcessor(job.getType()).onJobFinished(job); |
|||
} catch (Throwable e) { |
|||
log.error("Failed to process job update: {}", job, e); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void processJob(Job job) { |
|||
TenantId tenantId = job.getTenantId(); |
|||
JobId jobId = job.getId(); |
|||
try { |
|||
JobProcessor processor = getJobProcessor(job.getType()); |
|||
List<TaskResult> toReprocess = job.getConfiguration().getToReprocess(); |
|||
if (toReprocess == null) { |
|||
int tasksCount = processor.process(job, this::submitTask); |
|||
log.info("[{}][{}][{}] Submitted {} tasks", tenantId, jobId, job.getType(), tasksCount); |
|||
jobStatsService.reportAllTasksSubmitted(tenantId, jobId, tasksCount); |
|||
} else { |
|||
processor.reprocess(job, toReprocess, this::submitTask); |
|||
log.info("[{}][{}][{}] Submitted {} tasks for reprocessing", tenantId, jobId, job.getType(), toReprocess.size()); |
|||
} |
|||
} catch (Throwable e) { |
|||
log.error("[{}][{}][{}] Failed to submit tasks", tenantId, jobId, job.getType(), e); |
|||
jobService.markAsFailed(tenantId, jobId, e.getMessage()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void cancelJob(TenantId tenantId, JobId jobId) { |
|||
log.info("[{}][{}] Cancelling job", tenantId, jobId); |
|||
jobService.cancelJob(tenantId, jobId); |
|||
} |
|||
|
|||
@Override |
|||
public void reprocessJob(TenantId tenantId, JobId jobId) { |
|||
log.info("[{}][{}] Reprocessing job", tenantId, jobId); |
|||
Job job = jobService.findJobById(tenantId, jobId); |
|||
if (job.getStatus() != JobStatus.FAILED) { |
|||
throw new IllegalArgumentException("Job is not failed"); |
|||
} |
|||
|
|||
JobResult result = job.getResult(); |
|||
if (result.getGeneralError() != null) { |
|||
job.presetResult(); |
|||
} else { |
|||
List<TaskResult> taskFailures = result.getResults().stream() |
|||
.filter(taskResult -> !taskResult.isSuccess() && !taskResult.isDiscarded()) |
|||
.toList(); |
|||
if (result.getFailedCount() > taskFailures.size()) { |
|||
throw new IllegalArgumentException("Reprocessing not allowed since there are too many failures (more than " + taskFailures.size() + ")"); |
|||
} |
|||
result.setFailedCount(0); |
|||
result.setResults(result.getResults().stream() |
|||
.filter(TaskResult::isSuccess) |
|||
.toList()); |
|||
job.getConfiguration().setToReprocess(taskFailures); |
|||
} |
|||
job.getConfiguration().setTasksKey(UUID.randomUUID().toString()); |
|||
jobService.saveJob(tenantId, job); |
|||
} |
|||
|
|||
private void submitTask(Task<?> task) { |
|||
if (ObjectUtils.anyNull(task.getTenantId(), task.getJobId(), task.getKey())) { |
|||
throw new IllegalArgumentException("Task " + task + " missing required fields"); |
|||
} |
|||
|
|||
log.debug("[{}][{}] Submitting task: {}", task.getTenantId(), task.getJobId(), task); |
|||
TaskProto taskProto = TaskProto.newBuilder() |
|||
.setValue(JacksonUtil.toString(task)) |
|||
.build(); |
|||
|
|||
TbQueueProducer<TbProtoQueueMsg<TaskProto>> producer = taskProducers.get(task.getJobType()); |
|||
EntityId entityId = null; |
|||
if (queueConfig.getPartitioningStrategy().equals("entity")) { |
|||
entityId = task.getEntityId(); |
|||
} |
|||
if (entityId == null) { |
|||
entityId = task.getTenantId(); |
|||
} |
|||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TASK_PROCESSOR, task.getJobType().name(), task.getTenantId(), entityId); |
|||
producer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), taskProto), new TbQueueCallback() { |
|||
@Override |
|||
public void onSuccess(TbQueueMsgMetadata metadata) { |
|||
log.trace("Submitted task to {}: {}", tpi, taskProto); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.warn("Failed to submit task: {}", task, t); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private JobProcessor getJobProcessor(JobType jobType) { |
|||
return jobProcessors.get(jobType); |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void destroy() { |
|||
executor.shutdownNow(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,94 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.job.DummyJobConfiguration; |
|||
import org.thingsboard.server.common.data.job.Job; |
|||
import org.thingsboard.server.common.data.job.JobType; |
|||
import org.thingsboard.server.common.data.job.task.DummyTask; |
|||
import org.thingsboard.server.common.data.job.task.DummyTaskResult; |
|||
import org.thingsboard.server.common.data.job.task.DummyTaskResult.DummyTaskFailure; |
|||
import org.thingsboard.server.common.data.job.task.Task; |
|||
import org.thingsboard.server.common.data.job.task.TaskResult; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.function.Consumer; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class DummyJobProcessor implements JobProcessor { |
|||
|
|||
@Override |
|||
public int process(Job job, Consumer<Task<?>> taskConsumer) throws Exception { |
|||
DummyJobConfiguration configuration = job.getConfiguration(); |
|||
if (configuration.getGeneralError() != null) { |
|||
for (int number = 1; number <= configuration.getSubmittedTasksBeforeGeneralError(); number++) { |
|||
taskConsumer.accept(createTask(job, configuration, number, null, false)); |
|||
} |
|||
Thread.sleep(configuration.getTaskProcessingTimeMs() * (configuration.getSubmittedTasksBeforeGeneralError() / 2)); // sleeping so that some tasks are processed
|
|||
throw new RuntimeException(configuration.getGeneralError()); |
|||
} |
|||
|
|||
int taskNumber = 1; |
|||
for (int i = 0; i < configuration.getSuccessfulTasksCount(); i++) { |
|||
taskConsumer.accept(createTask(job, configuration, taskNumber, null, false)); |
|||
taskNumber++; |
|||
} |
|||
if (configuration.getErrors() != null) { |
|||
for (int i = 0; i < configuration.getFailedTasksCount(); i++) { |
|||
taskConsumer.accept(createTask(job, configuration, taskNumber, configuration.getErrors(), false)); |
|||
taskNumber++; |
|||
} |
|||
for (int i = 0; i < configuration.getPermanentlyFailedTasksCount(); i++) { |
|||
taskConsumer.accept(createTask(job, configuration, taskNumber, configuration.getErrors(), true)); |
|||
taskNumber++; |
|||
} |
|||
} |
|||
return configuration.getSuccessfulTasksCount() + configuration.getFailedTasksCount() + configuration.getPermanentlyFailedTasksCount(); |
|||
} |
|||
|
|||
@Override |
|||
public void reprocess(Job job, List<TaskResult> taskFailures, Consumer<Task<?>> taskConsumer) throws Exception { |
|||
for (TaskResult taskFailure : taskFailures) { |
|||
DummyTaskFailure failure = ((DummyTaskResult) taskFailure).getFailure(); |
|||
taskConsumer.accept(createTask(job, job.getConfiguration(), failure.getNumber(), failure.isFailAlways() ? |
|||
List.of(failure.getError()) : Collections.emptyList(), failure.isFailAlways())); |
|||
} |
|||
} |
|||
|
|||
private DummyTask createTask(Job job, DummyJobConfiguration configuration, int number, List<String> errors, boolean failAlways) { |
|||
return DummyTask.builder() |
|||
.tenantId(job.getTenantId()) |
|||
.jobId(job.getId()) |
|||
.key(configuration.getTasksKey()) |
|||
.retries(configuration.getRetries()) |
|||
.number(number) |
|||
.processingTimeMs(configuration.getTaskProcessingTimeMs()) |
|||
.errors(errors) |
|||
.failAlways(failAlways) |
|||
.processingTimeoutMs(configuration.getTaskProcessingTimeoutMs()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public JobType getType() { |
|||
return JobType.DUMMY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import org.thingsboard.server.common.data.job.Job; |
|||
import org.thingsboard.server.common.data.job.JobType; |
|||
import org.thingsboard.server.common.data.job.task.Task; |
|||
import org.thingsboard.server.common.data.job.task.TaskResult; |
|||
|
|||
import java.util.List; |
|||
import java.util.function.Consumer; |
|||
|
|||
public interface JobProcessor { |
|||
|
|||
int process(Job job, Consumer<Task<?>> taskConsumer) throws Exception; |
|||
|
|||
void reprocess(Job job, List<TaskResult> taskFailures, Consumer<Task<?>> taskConsumer) throws Exception; |
|||
|
|||
default void onJobFinished(Job job) {} |
|||
|
|||
JobType getType(); |
|||
|
|||
} |
|||
@ -0,0 +1,115 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import jakarta.annotation.PreDestroy; |
|||
import lombok.SneakyThrows; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.id.JobId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.job.JobStats; |
|||
import org.thingsboard.server.common.data.job.task.TaskResult; |
|||
import org.thingsboard.server.dao.job.JobService; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.JobStatsMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.settings.TasksQueueConfig; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
@TbCoreComponent |
|||
@Component |
|||
@Slf4j |
|||
public class JobStatsProcessor { |
|||
|
|||
private final JobService jobService; |
|||
private final TasksQueueConfig queueConfig; |
|||
private final QueueConsumerManager<TbProtoQueueMsg<JobStatsMsg>> jobStatsConsumer; |
|||
private final ExecutorService consumerExecutor; |
|||
|
|||
public JobStatsProcessor(JobService jobService, |
|||
TasksQueueConfig queueConfig, |
|||
TbCoreQueueFactory queueFactory) { |
|||
this.jobService = jobService; |
|||
this.queueConfig = queueConfig; |
|||
this.consumerExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("job-stats-consumer")); |
|||
this.jobStatsConsumer = QueueConsumerManager.<TbProtoQueueMsg<JobStatsMsg>>builder() |
|||
.name("job-stats") |
|||
.msgPackProcessor(this::processStats) |
|||
.pollInterval(queueConfig.getStatsPollInterval()) |
|||
.consumerCreator(queueFactory::createJobStatsConsumer) |
|||
.consumerExecutor(consumerExecutor) |
|||
.build(); |
|||
} |
|||
|
|||
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
|||
public void afterStartUp() { |
|||
jobStatsConsumer.subscribe(); |
|||
jobStatsConsumer.launch(); |
|||
} |
|||
|
|||
@SneakyThrows |
|||
private void processStats(List<TbProtoQueueMsg<JobStatsMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<JobStatsMsg>> consumer) { |
|||
Map<JobId, JobStats> stats = new HashMap<>(); |
|||
|
|||
for (TbProtoQueueMsg<JobStatsMsg> msg : msgs) { |
|||
JobStatsMsg statsMsg = msg.getValue(); |
|||
TenantId tenantId = TenantId.fromUUID(new UUID(statsMsg.getTenantIdMSB(), statsMsg.getTenantIdLSB())); |
|||
JobId jobId = new JobId(new UUID(statsMsg.getJobIdMSB(), statsMsg.getJobIdLSB())); |
|||
JobStats jobStats = stats.computeIfAbsent(jobId, __ -> new JobStats(tenantId, jobId)); |
|||
|
|||
if (statsMsg.hasTaskResult()) { |
|||
TaskResult taskResult = JacksonUtil.fromString(statsMsg.getTaskResult().getValue(), TaskResult.class); |
|||
jobStats.getTaskResults().add(taskResult); |
|||
} |
|||
if (statsMsg.hasTotalTasksCount()) { |
|||
jobStats.setTotalTasksCount(statsMsg.getTotalTasksCount()); |
|||
} |
|||
} |
|||
|
|||
stats.forEach((jobId, jobStats) -> { |
|||
TenantId tenantId = jobStats.getTenantId(); |
|||
try { |
|||
log.debug("[{}][{}] Processing job stats: {}", tenantId, jobId, stats); |
|||
jobService.processStats(tenantId, jobId, jobStats); |
|||
} catch (Exception e) { |
|||
log.error("[{}][{}] Failed to process job stats: {}", tenantId, jobId, jobStats, e); |
|||
} |
|||
}); |
|||
consumer.commit(); |
|||
|
|||
Thread.sleep(queueConfig.getStatsProcessingInterval()); |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void destroy() { |
|||
jobStatsConsumer.stop(); |
|||
consumerExecutor.shutdownNow(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,59 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job.task; |
|||
|
|||
import org.apache.commons.lang3.tuple.Pair; |
|||
import org.thingsboard.server.common.data.job.JobType; |
|||
import org.thingsboard.server.common.data.job.task.DummyTask; |
|||
import org.thingsboard.server.common.data.job.task.DummyTaskResult; |
|||
import org.thingsboard.server.common.data.job.task.Task; |
|||
import org.thingsboard.server.queue.task.TaskProcessor; |
|||
|
|||
import java.util.Map; |
|||
import java.util.concurrent.Future; |
|||
|
|||
public class DummyTaskProcessor extends TaskProcessor<DummyTask, DummyTaskResult> { |
|||
|
|||
@Override |
|||
public DummyTaskResult process(DummyTask task) throws Exception { |
|||
if (task.getProcessingTimeMs() > 0) { |
|||
Thread.sleep(task.getProcessingTimeMs()); |
|||
} |
|||
if (task.isFailAlways()) { |
|||
throw new RuntimeException(task.getErrors().get(0)); |
|||
} |
|||
if (task.getErrors() != null && task.getAttempt() <= task.getErrors().size()) { |
|||
String error = task.getErrors().get(task.getAttempt() - 1); |
|||
throw new RuntimeException(error); |
|||
} |
|||
return DummyTaskResult.success(task); |
|||
} |
|||
|
|||
@Override |
|||
public long getProcessingTimeout(DummyTask task) { |
|||
return task.getProcessingTimeoutMs() > 0 ? task.getProcessingTimeoutMs() : 2000; |
|||
} |
|||
|
|||
public Map<Object, Pair<Task<DummyTaskResult>, Future<DummyTaskResult>>> getCurrentTasks() { |
|||
return currentTasks; |
|||
} |
|||
|
|||
@Override |
|||
public JobType getJobType() { |
|||
return JobType.DUMMY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.notification.rule.trigger; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.notification.info.ResourcesShortageNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.ResourcesShortageTrigger; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.ResourcesShortageNotificationRuleTriggerConfig; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class ResourcesShortageTriggerProcessor implements NotificationRuleTriggerProcessor<ResourcesShortageTrigger, ResourcesShortageNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(ResourcesShortageTrigger trigger, ResourcesShortageNotificationRuleTriggerConfig triggerConfig) { |
|||
float usagePercent = trigger.getUsage() / 100.0f; |
|||
return switch (trigger.getResource()) { |
|||
case CPU -> usagePercent >= triggerConfig.getCpuThreshold(); |
|||
case RAM -> usagePercent >= triggerConfig.getRamThreshold(); |
|||
case STORAGE -> usagePercent >= triggerConfig.getStorageThreshold(); |
|||
}; |
|||
} |
|||
|
|||
@Override |
|||
public RuleOriginatedNotificationInfo constructNotificationInfo(ResourcesShortageTrigger trigger) { |
|||
return ResourcesShortageNotificationInfo.builder().resource(trigger.getResource().name()).usage(trigger.getUsage()).build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.RESOURCES_SHORTAGE; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,92 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.telemetry; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import com.google.common.util.concurrent.SettableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.kv.Aggregation; |
|||
import org.thingsboard.server.common.data.kv.AggregationParams; |
|||
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; |
|||
import org.thingsboard.server.common.data.kv.IntervalType; |
|||
import org.thingsboard.server.common.data.kv.ReadTsKvQuery; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
import org.thingsboard.server.service.security.AccessValidator; |
|||
import org.thingsboard.server.service.security.ValidationResult; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import java.util.List; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
public class DefaultTbTelemetryService implements TbTelemetryService { |
|||
|
|||
private final TimeseriesService tsService; |
|||
private final AccessValidator accessValidator; |
|||
|
|||
@Override |
|||
public ListenableFuture<List<TsKvEntry>> getTimeseries(EntityId entityId, List<String> keys, Long startTs, Long endTs, IntervalType intervalType, |
|||
Long interval, String timeZone, Integer limit, Aggregation agg, String orderBy, |
|||
Boolean useStrictDataTypes, SecurityUser currentUser) { |
|||
SettableFuture<List<TsKvEntry>> future = SettableFuture.create(); |
|||
accessValidator.validate(currentUser, Operation.READ_TELEMETRY, entityId, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(ValidationResult validationResult) { |
|||
try { |
|||
AggregationParams params; |
|||
if (Aggregation.NONE.equals(agg)) { |
|||
params = AggregationParams.none(); |
|||
} else if (intervalType == null || IntervalType.MILLISECONDS.equals(intervalType)) { |
|||
params = interval == 0L ? AggregationParams.none() : AggregationParams.milliseconds(agg, interval); |
|||
} else { |
|||
params = AggregationParams.calendar(agg, intervalType, timeZone); |
|||
} |
|||
List<ReadTsKvQuery> queries = keys.stream().map(key -> new BaseReadTsKvQuery(key, startTs, endTs, params, limit, orderBy)).collect(Collectors.toList()); |
|||
Futures.addCallback(tsService.findAll(currentUser.getTenantId(), entityId, queries), new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(List<TsKvEntry> result) { |
|||
future.set(result); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
future.setException(t); |
|||
} |
|||
}, MoreExecutors.directExecutor()); |
|||
} catch (Throwable e) { |
|||
onFailure(e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
future.setException(t); |
|||
} |
|||
}); |
|||
return future; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.telemetry; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.kv.Aggregation; |
|||
import org.thingsboard.server.common.data.kv.IntervalType; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface TbTelemetryService { |
|||
|
|||
ListenableFuture<List<TsKvEntry>> getTimeseries(EntityId entityId, |
|||
List<String> keys, |
|||
Long startTs, |
|||
Long endTs, |
|||
IntervalType intervalType, |
|||
Long interval, |
|||
String timeZone, |
|||
Integer limit, |
|||
Aggregation agg, |
|||
String orderBy, |
|||
Boolean useStrictDataTypes, |
|||
SecurityUser currentUser) throws ThingsboardException; |
|||
|
|||
} |
|||
@ -0,0 +1,504 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import lombok.SneakyThrows; |
|||
import org.junit.After; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.mockito.Mockito; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.test.mock.mockito.SpyBean; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.thingsboard.rule.engine.api.JobManager; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.JobId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.job.DummyJobConfiguration; |
|||
import org.thingsboard.server.common.data.job.Job; |
|||
import org.thingsboard.server.common.data.job.JobFilter; |
|||
import org.thingsboard.server.common.data.job.JobResult; |
|||
import org.thingsboard.server.common.data.job.JobStatus; |
|||
import org.thingsboard.server.common.data.job.JobType; |
|||
import org.thingsboard.server.common.data.job.task.DummyTaskResult; |
|||
import org.thingsboard.server.common.data.job.task.DummyTaskResult.DummyTaskFailure; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.controller.AbstractControllerTest; |
|||
import org.thingsboard.server.dao.job.JobDao; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.queue.task.JobStatsService; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Comparator; |
|||
import java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.Mockito.doAnswer; |
|||
import static org.mockito.Mockito.never; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
|
|||
@DaoSqlTest |
|||
@TestPropertySource(properties = { |
|||
"queue.tasks.stats.processing_interval=0" |
|||
}) |
|||
public class JobManagerTest extends AbstractControllerTest { |
|||
|
|||
@Autowired |
|||
private JobManager jobManager; |
|||
|
|||
@SpyBean |
|||
private TestTaskProcessor taskProcessor; |
|||
|
|||
@SpyBean |
|||
private JobStatsService jobStatsService; |
|||
|
|||
@Autowired |
|||
private JobDao jobDao; |
|||
|
|||
private TenantId tenantId; |
|||
private Device jobEntity; |
|||
|
|||
@Before |
|||
public void setUp() throws Exception { |
|||
loginTenantAdmin(); |
|||
tenantId = super.tenantId; |
|||
jobEntity = createDevice("Test", "Test"); |
|||
} |
|||
|
|||
@After |
|||
public void tearDown() throws Exception { |
|||
} |
|||
|
|||
@Test |
|||
public void testSubmitJob_allTasksSuccessful() { |
|||
int tasksCount = 5; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(tasksCount) |
|||
.taskProcessingTimeMs(1000) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.RUNNING); |
|||
assertThat(job.getResult().getSuccessfulCount()).isBetween(1, tasksCount - 1); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(tasksCount); |
|||
}); |
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.COMPLETED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isEqualTo(tasksCount); |
|||
assertThat(job.getResult().getResults()).isEmpty(); |
|||
assertThat(job.getResult().getCompletedCount()).isEqualTo(tasksCount); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testSubmitJob_someTasksPermanentlyFailed() { |
|||
int successfulTasks = 3; |
|||
int failedTasks = 2; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(successfulTasks) |
|||
.failedTasksCount(failedTasks) |
|||
.errors(List.of("error1", "error2", "error3")) |
|||
.retries(2) |
|||
.taskProcessingTimeMs(100) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
JobResult jobResult = job.getResult(); |
|||
assertThat(jobResult.getSuccessfulCount()).isEqualTo(successfulTasks); |
|||
assertThat(jobResult.getFailedCount()).isEqualTo(failedTasks); |
|||
assertThat(jobResult.getTotalCount()).isEqualTo(successfulTasks + failedTasks); |
|||
assertThat(getFailures(jobResult)).hasSize(2).allSatisfy(failure -> { |
|||
assertThat(failure.getError()).isEqualTo("error3"); // last error
|
|||
}); |
|||
assertThat(jobResult.getCompletedCount()).isEqualTo(jobResult.getTotalCount()); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testSubmitJob_taskTimeout() { |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(1) |
|||
.taskProcessingTimeMs(5000) // bigger than DummyTaskProcessor.getTaskProcessingTimeout()
|
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
JobResult jobResult = job.getResult(); |
|||
assertThat(jobResult.getFailedCount()).isEqualTo(1); |
|||
assertThat(((DummyTaskResult) jobResult.getResults().get(0)).getFailure().getError()).isEqualTo("Timeout after 2000 ms"); // last error
|
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testCancelJob_whileRunning() throws Exception { |
|||
int tasksCount = 100; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(tasksCount) |
|||
.taskProcessingTimeMs(100) |
|||
.build()).getId(); |
|||
|
|||
Thread.sleep(500); |
|||
cancelJob(jobId); |
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.CANCELLED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isBetween(1, tasksCount - 1); |
|||
assertThat(job.getResult().getDiscardedCount()).isBetween(1, tasksCount - 1); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(tasksCount); |
|||
assertThat(job.getResult().getCompletedCount()).isEqualTo(tasksCount); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testCancelJob_whileTaskRunning() throws Exception { |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(1) |
|||
.taskProcessingTimeMs(TimeUnit.HOURS.toMillis(1)) |
|||
.taskProcessingTimeoutMs(TimeUnit.HOURS.toMillis(1)) |
|||
.build()).getId(); |
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
assertThat(taskProcessor.getCurrentTasks()).isNotEmpty(); |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.RUNNING); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(1); |
|||
assertThat(job.getResult().getCompletedCount()).isZero(); |
|||
}); |
|||
|
|||
cancelJob(jobId); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.CANCELLED); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(1); |
|||
assertThat(job.getResult().getDiscardedCount()).isEqualTo(1); |
|||
assertThat(job.getResult().getFailedCount()).isZero(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testCancelJob_simulateTaskProcessorRestart() throws Exception { |
|||
int tasksCount = 10; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(tasksCount) |
|||
.taskProcessingTimeMs(500) |
|||
.build()).getId(); |
|||
|
|||
// simulate cancelled jobs are forgotten
|
|||
AtomicInteger cancellationRenotifyAttempt = new AtomicInteger(0); |
|||
doAnswer(inv -> { |
|||
if (cancellationRenotifyAttempt.incrementAndGet() >= 5) { |
|||
inv.callRealMethod(); |
|||
} |
|||
return null; |
|||
}).when(taskProcessor).addToDiscarded(any()); // ignoring cancellation event,
|
|||
cancelJob(jobId); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.CANCELLED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isBetween(1, tasksCount - 1); |
|||
assertThat(job.getResult().getDiscardedCount()).isBetween(1, tasksCount - 1); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(tasksCount); |
|||
assertThat(job.getResult().getCompletedCount()).isEqualTo(tasksCount); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenTenantIsDeleted_thenCancelAllTheJobs() throws Exception { |
|||
loginSysAdmin(); |
|||
createDifferentTenant(); |
|||
|
|||
this.tenantId = this.differentTenantId; |
|||
submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(1000) |
|||
.taskProcessingTimeMs(500) |
|||
.build()); |
|||
|
|||
Thread.sleep(2000); |
|||
deleteDifferentTenant(); |
|||
Mockito.reset(jobStatsService); |
|||
|
|||
Thread.sleep(3000); |
|||
verify(jobStatsService, never()).reportTaskResult(any(), any(), any()); |
|||
assertThat(jobDao.findByTenantIdAndFilter(tenantId, JobFilter.builder().build(), new PageLink(100)).getData()).isEmpty(); |
|||
} |
|||
|
|||
@Test |
|||
public void testSubmitMultipleJobs() throws Exception { |
|||
int tasksCount = 3; |
|||
int jobsCount = 3; |
|||
for (int i = 1; i <= jobsCount; i++) { |
|||
submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(tasksCount) |
|||
.taskProcessingTimeMs(1000) |
|||
.build(), "test-job-" + i); |
|||
} |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
List<Job> jobs = findJobs(List.of(JobType.DUMMY), List.of(jobEntity.getUuidId())); |
|||
assertThat(jobs).hasSize(jobsCount); |
|||
Job firstJob = jobs.get(2); // ordered by createdTime descending
|
|||
assertThat(firstJob.getStatus()).isEqualTo(JobStatus.RUNNING); |
|||
Job secondJob = jobs.get(1); |
|||
assertThat(secondJob.getStatus()).isEqualTo(JobStatus.QUEUED); |
|||
Job thirdJob = jobs.get(0); |
|||
assertThat(thirdJob.getStatus()).isEqualTo(JobStatus.QUEUED); |
|||
}); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
List<Job> jobs = findJobs(); |
|||
for (Job job : jobs) { |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.COMPLETED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isEqualTo(tasksCount); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(tasksCount); |
|||
assertThat(job.getEntityId()).isEqualTo(jobEntity.getId()); |
|||
assertThat(job.getEntityName()).isEqualTo(jobEntity.getName()); |
|||
} |
|||
}); |
|||
|
|||
doDelete("/api/device/" + jobEntity.getId()).andExpect(status().isOk()); |
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
assertThat(findJobs(List.of(JobType.DUMMY), List.of(jobEntity.getUuidId()))).isEmpty(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testCancelQueuedJob() throws Exception { |
|||
int tasksCount = 3; |
|||
int jobsCount = 3; |
|||
List<JobId> jobIds = new ArrayList<>(); |
|||
for (int i = 1; i <= jobsCount; i++) { |
|||
Job job = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(tasksCount) |
|||
.taskProcessingTimeMs(1000) |
|||
.build(), "test-job-" + i); |
|||
jobIds.add(job.getId()); |
|||
} |
|||
|
|||
for (int i = 1; i < jobIds.size(); i++) { |
|||
cancelJob(jobIds.get(i)); |
|||
} |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
List<Job> jobs = findJobs(); |
|||
|
|||
Job firstJob = jobs.get(2); |
|||
assertThat(firstJob.getStatus()).isEqualTo(JobStatus.COMPLETED); |
|||
assertThat(firstJob.getResult().getSuccessfulCount()).isEqualTo(tasksCount); |
|||
assertThat(firstJob.getResult().getTotalCount()).isEqualTo(tasksCount); |
|||
|
|||
Job secondJob = jobs.get(1); |
|||
assertThat(secondJob.getStatus()).isEqualTo(JobStatus.CANCELLED); |
|||
assertThat(secondJob.getResult().getCompletedCount()).isZero(); |
|||
|
|||
Job thirdJob = jobs.get(0); |
|||
assertThat(thirdJob.getStatus()).isEqualTo(JobStatus.CANCELLED); |
|||
assertThat(thirdJob.getResult().getCompletedCount()).isZero(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testSubmitJob_generalError() { |
|||
int submittedTasks = 100; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.generalError("Some error while submitting tasks") |
|||
.submittedTasksBeforeGeneralError(submittedTasks) |
|||
.taskProcessingTimeMs(10) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isBetween(1, submittedTasks); |
|||
assertThat(job.getResult().getDiscardedCount()).isZero(); |
|||
assertThat(job.getResult().getTotalCount()).isNull(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testSubmitJob_immediateGeneralError() { |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.generalError("Some error while submitting tasks") |
|||
.submittedTasksBeforeGeneralError(0) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isZero(); |
|||
assertThat(job.getResult().getDiscardedCount()).isZero(); |
|||
assertThat(job.getResult().getFailedCount()).isZero(); |
|||
assertThat(job.getResult().getTotalCount()).isNull(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testReprocessJob_generalError() throws Exception { |
|||
int submittedTasks = 100; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.generalError("Some error while submitting tasks") |
|||
.submittedTasksBeforeGeneralError(submittedTasks) |
|||
.taskProcessingTimeMs(10) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
assertThat(job.getResult().getGeneralError()).isEqualTo("Some error while submitting tasks"); |
|||
}); |
|||
|
|||
Job savedJob = jobDao.findById(tenantId, jobId.getId()); |
|||
DummyJobConfiguration configuration = savedJob.getConfiguration(); |
|||
configuration.setGeneralError(null); |
|||
configuration.setSuccessfulTasksCount(submittedTasks); |
|||
jobDao.save(tenantId, savedJob); |
|||
|
|||
reprocessJob(jobId); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.COMPLETED); |
|||
assertThat(job.getResult().getGeneralError()).isNull(); |
|||
assertThat(job.getResult().getSuccessfulCount()).isEqualTo(submittedTasks); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(submittedTasks); |
|||
assertThat(job.getResult().getFailedCount()).isZero(); |
|||
assertThat(job.getResult().getDiscardedCount()).isZero(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testReprocessJob() throws Exception { |
|||
int successfulTasks = 3; |
|||
int failedTasks = 2; |
|||
int totalTasksCount = successfulTasks + failedTasks; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(successfulTasks) |
|||
.failedTasksCount(failedTasks) |
|||
.errors(List.of("error")) |
|||
.taskProcessingTimeMs(100) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
JobResult jobResult = job.getResult(); |
|||
assertThat(jobResult.getSuccessfulCount()).isEqualTo(successfulTasks); |
|||
assertThat(jobResult.getFailedCount()).isEqualTo(failedTasks); |
|||
|
|||
List<DummyTaskFailure> failures = getFailures(jobResult); |
|||
for (int i = 0, taskNumber = successfulTasks + 1; taskNumber <= totalTasksCount; i++, taskNumber++) { |
|||
DummyTaskFailure failure = failures.get(i); |
|||
assertThat(failure.getNumber()).isEqualTo(taskNumber); |
|||
assertThat(failure.getError()).isEqualTo("error"); |
|||
} |
|||
}); |
|||
|
|||
reprocessJob(jobId); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.COMPLETED); |
|||
assertThat(job.getResult().getSuccessfulCount()).isEqualTo(totalTasksCount); |
|||
assertThat(job.getResult().getFailedCount()).isZero(); |
|||
assertThat(job.getResult().getTotalCount()).isEqualTo(totalTasksCount); |
|||
assertThat(job.getResult().getResults()).isEmpty(); |
|||
assertThat(job.getConfiguration().getToReprocess()).isNullOrEmpty(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testReprocessJob_somePermanentlyFailed() throws Exception { |
|||
int successfulTasks = 3; |
|||
int failedTasks = 2; |
|||
int permanentlyFailedTasks = 1; |
|||
int totalTasksCount = successfulTasks + failedTasks + permanentlyFailedTasks; |
|||
JobId jobId = submitJob(DummyJobConfiguration.builder() |
|||
.successfulTasksCount(successfulTasks) |
|||
.failedTasksCount(failedTasks) |
|||
.permanentlyFailedTasksCount(permanentlyFailedTasks) |
|||
.errors(List.of("error")) |
|||
.taskProcessingTimeMs(100) |
|||
.build()).getId(); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
JobResult jobResult = job.getResult(); |
|||
assertThat(jobResult.getSuccessfulCount()).isEqualTo(successfulTasks); |
|||
assertThat(jobResult.getFailedCount()).isEqualTo(failedTasks + permanentlyFailedTasks); |
|||
assertThat(jobResult.getTotalCount()).isEqualTo(totalTasksCount); |
|||
|
|||
List<DummyTaskFailure> failures = getFailures(jobResult); |
|||
for (int i = 0, taskNumber = successfulTasks + 1; taskNumber <= totalTasksCount; i++, taskNumber++) { |
|||
DummyTaskFailure failure = failures.get(i); |
|||
assertThat(failure.getNumber()).isEqualTo(taskNumber); |
|||
assertThat(failure.getError()).isEqualTo("error"); |
|||
} |
|||
}); |
|||
|
|||
reprocessJob(jobId); |
|||
|
|||
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
Job job = findJobById(jobId); |
|||
assertThat(job.getStatus()).isEqualTo(JobStatus.FAILED); |
|||
JobResult jobResult = job.getResult(); |
|||
assertThat(jobResult.getSuccessfulCount()).isEqualTo(successfulTasks + failedTasks); |
|||
assertThat(jobResult.getFailedCount()).isEqualTo(permanentlyFailedTasks); |
|||
assertThat(jobResult.getTotalCount()).isEqualTo(totalTasksCount); |
|||
|
|||
List<DummyTaskFailure> failures = getFailures(jobResult); |
|||
for (int i = 0, taskNumber = successfulTasks + failedTasks + 1; taskNumber <= totalTasksCount; i++, taskNumber++) { |
|||
DummyTaskFailure failure = failures.get(i); |
|||
assertThat(failure.getNumber()).isEqualTo(taskNumber); |
|||
assertThat(failure.getError()).isEqualTo("error"); |
|||
assertThat(failure.isFailAlways()).isTrue(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private Job submitJob(DummyJobConfiguration configuration) { |
|||
return submitJob(configuration, "test-job"); |
|||
} |
|||
|
|||
@SneakyThrows |
|||
private Job submitJob(DummyJobConfiguration configuration, String key) { |
|||
return jobManager.submitJob(Job.builder() |
|||
.tenantId(tenantId) |
|||
.type(JobType.DUMMY) |
|||
.key(key) |
|||
.entityId(jobEntity.getId()) |
|||
.configuration(configuration) |
|||
.build()).get(); |
|||
} |
|||
|
|||
private List<DummyTaskFailure> getFailures(JobResult jobResult) { |
|||
return jobResult.getResults().stream() |
|||
.map(taskResult -> ((DummyTaskResult) taskResult).getFailure()) |
|||
.sorted(Comparator.comparingInt(DummyTaskFailure::getNumber)) |
|||
.toList(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
|
|||
@DaoSqlTest |
|||
@TestPropertySource(properties = { |
|||
"queue.tasks.stats.processing_interval=0", |
|||
"queue.tasks.partitioning_strategy=entity", |
|||
"queue.tasks.partitions_per_type=DUMMY:100;DUMMY:50" |
|||
}) |
|||
public class JobManagerTest_EntityPartitioningStrategy extends JobManagerTest { |
|||
|
|||
/* |
|||
* Some tests are overridden because they are based on |
|||
* tenant partitioning strategy (subsequent tasks processing within a tenant) |
|||
* */ |
|||
|
|||
@Override |
|||
public void testCancelJob_simulateTaskProcessorRestart() throws Exception { |
|||
} |
|||
|
|||
@Override |
|||
public void testSubmitJob_generalError() { |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.job; |
|||
|
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.service.job.task.DummyTaskProcessor; |
|||
|
|||
@Component |
|||
public class TestTaskProcessor extends DummyTaskProcessor { |
|||
} |
|||
@ -0,0 +1,130 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.service.subscription; |
|||
|
|||
import org.apache.commons.lang3.RandomStringUtils; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.testcontainers.shaded.org.apache.commons.lang3.RandomUtils; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
|||
import org.thingsboard.server.common.data.kv.LongDataEntry; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.query.EntityData; |
|||
import org.thingsboard.server.common.data.query.EntityKeyType; |
|||
import org.thingsboard.server.common.data.query.TsValue; |
|||
import org.thingsboard.server.service.ws.WebSocketService; |
|||
import org.thingsboard.server.service.ws.WebSocketSessionRef; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate; |
|||
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.BDDMockito.then; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class TbEntityDataSubCtxTest { |
|||
|
|||
private final DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
|
|||
private final Integer cmdId = RandomUtils.nextInt(); |
|||
private final Integer subscriptionId = RandomUtils.nextInt(); |
|||
private final String serviceId = RandomStringUtils.randomAlphanumeric(10); |
|||
private final String sessionId = RandomStringUtils.randomAlphanumeric(10); |
|||
|
|||
private final int maxEntitiesPerDataSubscription = 100; |
|||
|
|||
private TbEntityDataSubCtx subCtx; |
|||
@Mock |
|||
private WebSocketService webSocketService; |
|||
@Mock |
|||
private WebSocketSessionRef webSocketSessionRef; |
|||
|
|||
@BeforeEach |
|||
public void setUp() { |
|||
when(webSocketSessionRef.getSessionId()).thenReturn(sessionId); |
|||
subCtx = new TbEntityDataSubCtx(serviceId, webSocketService, mock(), mock(), mock(), mock(), webSocketSessionRef, cmdId, maxEntitiesPerDataSubscription); |
|||
|
|||
Map<Integer, EntityId> subToEntityIdMap = new HashMap<>(); |
|||
subToEntityIdMap.put(subscriptionId, deviceId); |
|||
ReflectionTestUtils.setField(subCtx, "subToEntityIdMap", subToEntityIdMap); |
|||
|
|||
long now = System.currentTimeMillis(); |
|||
long oldTs = now - 1_000_000; |
|||
|
|||
Map<String, TsValue> latestCtxValues = new HashMap<>(); |
|||
latestCtxValues.put("key", new TsValue(oldTs, "15")); |
|||
Map<EntityKeyType, Map<String, TsValue>> latest = new HashMap<>(); |
|||
latest.put(EntityKeyType.TIME_SERIES, latestCtxValues); |
|||
|
|||
EntityData entityData = new EntityData(); |
|||
entityData.setEntityId(deviceId); |
|||
entityData.setLatest(latest); |
|||
|
|||
PageData<EntityData> data = new PageData<>(List.of(entityData), 1, 1, true); |
|||
ReflectionTestUtils.setField(subCtx, "data", data); |
|||
} |
|||
|
|||
@Test |
|||
public void testSendLatestWsMsg() { |
|||
long ts = System.currentTimeMillis(); |
|||
List<TsKvEntry> telemetry = List.of( |
|||
new BasicTsKvEntry(ts - 50000, new LongDataEntry("key", 42L), 34L), |
|||
new BasicTsKvEntry(ts - 20000, new LongDataEntry("key", 17L), 78L) |
|||
); |
|||
|
|||
TelemetrySubscriptionUpdate subUpdate = new TelemetrySubscriptionUpdate(subscriptionId, telemetry); |
|||
|
|||
subCtx.sendWsMsg(sessionId, subUpdate, EntityKeyType.TIME_SERIES, true); |
|||
|
|||
Map<EntityKeyType, Map<String, TsValue>> expectedLatest = new HashMap<>(); |
|||
Map<String, TsValue> expectedLatestCtxValues = new HashMap<>(); |
|||
expectedLatestCtxValues.put("key", new TsValue(ts - 20000, "17")); // use latest telemetry
|
|||
expectedLatest.put(EntityKeyType.TIME_SERIES, expectedLatestCtxValues); |
|||
|
|||
EntityData expectedEntityData = new EntityData(); |
|||
expectedEntityData.setEntityId(deviceId); |
|||
expectedEntityData.setLatest(expectedLatest); |
|||
|
|||
List<EntityData> expected = List.of(expectedEntityData); |
|||
|
|||
ArgumentCaptor<CmdUpdate> cmdUpdateCaptor = ArgumentCaptor.forClass(CmdUpdate.class); |
|||
then(webSocketService).should().sendUpdate(eq(sessionId), cmdUpdateCaptor.capture()); |
|||
CmdUpdate cmdUpdate = cmdUpdateCaptor.getValue(); |
|||
assertThat(cmdUpdate).isInstanceOf(EntityDataUpdate.class); |
|||
EntityDataUpdate entityDataUpdate = (EntityDataUpdate) cmdUpdate; |
|||
assertThat(entityDataUpdate.getCmdId()).isEqualTo(cmdId); |
|||
assertThat(entityDataUpdate.getData()).isNull(); |
|||
assertThat(entityDataUpdate.getUpdate()).isEqualTo(expected); |
|||
assertThat(entityDataUpdate.getAllowedEntities()).isEqualTo(maxEntitiesPerDataSubscription); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.dao.job; |
|||
|
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.JobId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.job.Job; |
|||
import org.thingsboard.server.common.data.job.JobFilter; |
|||
import org.thingsboard.server.common.data.job.JobStats; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.dao.entity.EntityDaoService; |
|||
|
|||
public interface JobService extends EntityDaoService { |
|||
|
|||
Job saveJob(TenantId tenantId, Job job); |
|||
|
|||
Job findJobById(TenantId tenantId, JobId jobId); |
|||
|
|||
void cancelJob(TenantId tenantId, JobId jobId); |
|||
|
|||
void markAsFailed(TenantId tenantId, JobId jobId, String error); |
|||
|
|||
void processStats(TenantId tenantId, JobId jobId, JobStats jobStats); |
|||
|
|||
PageData<Job> findJobsByFilter(TenantId tenantId, JobFilter filter, PageLink pageLink); |
|||
|
|||
Job findLatestJobByKey(TenantId tenantId, String key); |
|||
|
|||
void deleteJob(TenantId tenantId, JobId jobId); |
|||
|
|||
int deleteJobsByEntityId(TenantId tenantId, EntityId entityId); |
|||
|
|||
} |
|||
@ -0,0 +1,18 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.edqs; |
|||
|
|||
public interface EdqsObjectKey {} |
|||
@ -0,0 +1,75 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.edqs; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import lombok.Getter; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.Setter; |
|||
import org.apache.commons.lang3.BooleanUtils; |
|||
|
|||
@Getter |
|||
@NoArgsConstructor |
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
public class EdqsState { |
|||
|
|||
private Boolean edqsReady; |
|||
@Setter |
|||
private EdqsSyncStatus syncStatus; |
|||
@Setter |
|||
private EdqsApiMode apiMode; |
|||
|
|||
public boolean updateEdqsReady(boolean ready) { |
|||
boolean changed = BooleanUtils.toBooleanDefaultIfNull(this.edqsReady, false) != ready; |
|||
this.edqsReady = ready; |
|||
return changed; |
|||
} |
|||
|
|||
@JsonIgnore |
|||
public boolean isApiReady() { |
|||
return edqsReady && syncStatus == EdqsSyncStatus.FINISHED; |
|||
} |
|||
|
|||
@JsonIgnore |
|||
public boolean isApiEnabled() { |
|||
return apiMode != null && (apiMode == EdqsApiMode.ENABLED || apiMode == EdqsApiMode.AUTO_ENABLED); |
|||
} |
|||
|
|||
@Override |
|||
public String toString() { |
|||
return '[' + |
|||
"EDQS ready: " + edqsReady + |
|||
", sync status: " + syncStatus + |
|||
", API mode: " + apiMode + |
|||
']'; |
|||
} |
|||
|
|||
public enum EdqsSyncStatus { |
|||
REQUESTED, |
|||
STARTED, |
|||
FINISHED, |
|||
FAILED |
|||
} |
|||
|
|||
public enum EdqsApiMode { |
|||
ENABLED, |
|||
AUTO_ENABLED, |
|||
DISABLED, |
|||
AUTO_DISABLED |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.id; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonCreator; |
|||
import com.fasterxml.jackson.annotation.JsonProperty; |
|||
import io.swagger.v3.oas.annotations.media.Schema; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
public class JobId extends UUIDBased implements EntityId { |
|||
|
|||
@JsonCreator |
|||
public JobId(@JsonProperty("id") UUID id) { |
|||
super(id); |
|||
} |
|||
|
|||
@Schema(requiredMode = Schema.RequiredMode.REQUIRED, description = "string", example = "JOB", allowableValues = "JOB") |
|||
@Override |
|||
public EntityType getEntityType() { |
|||
return EntityType.JOB; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.job; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
@Builder |
|||
@ToString(callSuper = true) |
|||
public class DummyJobConfiguration extends JobConfiguration { |
|||
|
|||
private long taskProcessingTimeMs; |
|||
private int successfulTasksCount; |
|||
private int failedTasksCount; |
|||
private int permanentlyFailedTasksCount; |
|||
private List<String> errors; |
|||
private int retries; |
|||
private long taskProcessingTimeoutMs; |
|||
|
|||
private String generalError; |
|||
private int submittedTasksBeforeGeneralError; |
|||
|
|||
@Override |
|||
public JobType getType() { |
|||
return JobType.DUMMY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.job; |
|||
|
|||
public class DummyJobResult extends JobResult { |
|||
|
|||
@Override |
|||
public JobType getJobType() { |
|||
return JobType.DUMMY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,85 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.job; |
|||
|
|||
import jakarta.validation.Valid; |
|||
import jakarta.validation.constraints.NotBlank; |
|||
import jakarta.validation.constraints.NotNull; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.BaseData; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.HasTenantId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.JobId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
public class Job extends BaseData<JobId> implements HasTenantId { |
|||
|
|||
@NotNull |
|||
private TenantId tenantId; |
|||
@NotNull |
|||
private JobType type; |
|||
@NotBlank |
|||
private String key; |
|||
@NotNull |
|||
private EntityId entityId; |
|||
private String entityName; // read-only
|
|||
@NotNull |
|||
private JobStatus status; |
|||
@NotNull |
|||
@Valid |
|||
private JobConfiguration configuration; |
|||
@NotNull |
|||
private JobResult result; |
|||
|
|||
public static final Set<EntityType> SUPPORTED_ENTITY_TYPES = Set.of( |
|||
EntityType.DEVICE, EntityType.ASSET, EntityType.DEVICE_PROFILE, EntityType.ASSET_PROFILE |
|||
); |
|||
|
|||
@Builder(toBuilder = true) |
|||
public Job(TenantId tenantId, JobType type, String key, EntityId entityId, JobConfiguration configuration) { |
|||
this.tenantId = tenantId; |
|||
this.type = type; |
|||
this.key = key; |
|||
this.entityId = entityId; |
|||
this.configuration = configuration; |
|||
this.configuration.setTasksKey(UUID.randomUUID().toString()); |
|||
presetResult(); |
|||
} |
|||
|
|||
public void presetResult() { |
|||
this.result = switch (type) { |
|||
case DUMMY -> new DummyJobResult(); |
|||
}; |
|||
} |
|||
|
|||
@SuppressWarnings("unchecked") |
|||
public <C extends JobConfiguration> C getConfiguration() { |
|||
return (C) configuration; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* 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 org.thingsboard.server.common.data.job; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import com.fasterxml.jackson.annotation.JsonSubTypes; |
|||
import com.fasterxml.jackson.annotation.JsonSubTypes.Type; |
|||
import com.fasterxml.jackson.annotation.JsonTypeInfo; |
|||
import jakarta.validation.constraints.NotBlank; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.job.task.TaskResult; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.List; |
|||
|
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") |
|||
@JsonSubTypes({ |
|||
@Type(name = "DUMMY", value = DummyJobConfiguration.class), |
|||
}) |
|||
@Data |
|||
public abstract class JobConfiguration implements Serializable { |
|||
|
|||
@NotBlank |
|||
private String tasksKey; // internal
|
|||
private List<TaskResult> toReprocess; // internal
|
|||
|
|||
@JsonIgnore |
|||
public abstract JobType getType(); |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue