dashevchenko
2 years ago
318 changed files with 8632 additions and 2499 deletions
File diff suppressed because one or more lines are too long
@ -0,0 +1,29 @@ |
|||
{ |
|||
"fqn": "charts.bars", |
|||
"name": "Bars", |
|||
"deprecated": true, |
|||
"image": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAMgAAACgCAMAAAB+IdObAAAA8FBMVEUhlvNMr1Bqamp5eXl7e3t8fHx9fX1+fn5/f3+AgICCgoKDg4OEhISGhoaHh4eKioqMjIyNjY2Ojo6QkJCRkZGSkpKWlpaXl5ebm5udnZ2enp6goKChoaGkpKSnp6epqamsrKyurq6xsbGzs7O1tbW2tra3t7e4uLi7u7u9vb3BwcHCwsLDw8PGxsbKysrNzc3Ozs7R0dHS0tLT09PZ2dna2trc3Nzd3d3e3t7g4ODh4eHj4+Pk5OTm5ubn5+fo6Ojp6enu7u7w8PDz8/P0Qzb09PT29vb39/f5+fn6+vr7+/v8/Pz9/f3+/v7/wQf///+dc+aLAAAAAWJLR0RPbmZBSQAAAcFJREFUeNrt3ds2AgEYhuHsaSOZbAvZi0r2YYjCJOW7/7txZhkcDNbM6h/vdwfPmlX/ybtmEorJErGCeJLadz3rkKPpZamaLTp925DHdFvSpKelU9uQ/cLKhtcdk7YqtiHruevtojch7ZZtQ0o1dcdfRqXNqm1IbVU3OaVamm/YhvQW5zIXOknnC5JUt7qEpE5fUv/5HVePy2UHAgQIECBAgAABAgQIECBxgrwGHBAgQIAAAQIECBAgQIAAAQIECJC/QRIBN0iQ+66voDMLuRp2fQWdVUhvNun6CjqrkJ0Dx/UVdEYhzXzfcX0FnVFIrlSZ2mx/LOiMQuqHh6k972NBZ/fv13G/KeiCQkIu4358EL8UdBafSGwuOxAgQIAAAQIECJDYQB4CDggQIECAAAECBAgQIECAAAECBAgQIECA/BrSufn0DjqjkEZmLXkWaUEXEmThXMeFSAu68H4j5b1IC7rQILfZTqQFXViQ1nRTL1EWdCFBnmYuJUVZ0IUEWR1xHKcXZUEXFDLwBR2XHQgQIECAAAEC5H9ChgIOCBAgQIAAAQIECBAgQIAAAQIECBAgQIAAAQIECBAgxiBmv+L6Bl9pkxYph15gAAAAAElFTkSuQmCC", |
|||
"description": "Displays latest values of the attributes or time-series data for multiple entities as separate bars.", |
|||
"descriptor": { |
|||
"type": "latest", |
|||
"sizeX": 7, |
|||
"sizeY": 5, |
|||
"resources": [ |
|||
{ |
|||
"url": "https://cdnjs.cloudflare.com/ajax/libs/Chart.js/2.3.0/Chart.min.js" |
|||
} |
|||
], |
|||
"templateHtml": "<canvas id=\"barChart\"></canvas>\n", |
|||
"templateCss": "", |
|||
"controllerScript": "self.onInit = function() {\n $scope = self.ctx.$scope;\n utils = $scope.$injector.get(self.ctx.servicesMap.get('utils'));\n settings = utils.deepClone(self.ctx.settings) || {};\n settings.showTooltip = utils.defaultValue(settings.showTooltip, true);\n \n Chart.defaults.global.tooltips.enabled = settings.showTooltip;\n \n var barData = {\n labels: [],\n datasets: []\n };\n \n for (var i = 0; i < self.ctx.datasources.length; i++) {\n var datasource = self.ctx.datasources[i];\n for (var d = 0; d < datasource.dataKeys.length; d++) {\n var dataKey = datasource.dataKeys[d];\n var units = dataKey.units && dataKey.units.length ? dataKey.units : self.ctx.units;\n units = units ? (' (' + units + ')') : '';\n var dataset = {\n label: dataKey.label + units,\n data: [0],\n backgroundColor: [dataKey.color],\n borderColor: [dataKey.color],\n borderWidth: 1\n }\n barData.datasets.push(dataset);\n }\n }\n\n var ctx = $('#barChart', self.ctx.$container);\n self.ctx.chart = new Chart(ctx, {\n type: 'bar',\n data: barData,\n options: {\n responsive: false,\n maintainAspectRatio: false,\n scales: {\n yAxes: [{\n ticks: {\n beginAtZero:true\n }\n }]\n }\n }\n });\n \n self.onResize();\n}\n\nself.onDataUpdated = function() {\n var c = 0;\n for (var i = 0; i < self.ctx.chart.data.datasets.length; i++) {\n var dataset = self.ctx.chart.data.datasets[i];\n var cellData = self.ctx.data[i]; \n if (cellData.data.length > 0) {\n var decimals;\n if (typeof cellData.dataKey.decimals !== 'undefined' \n && cellData.dataKey.decimals !== null ) {\n decimals = cellData.dataKey.decimals; \n } else {\n decimals = self.ctx.decimals;\n }\n var tvPair = cellData.data[cellData.data.length - 1];\n var value = self.ctx.utils.formatValue(tvPair[1], decimals);\n dataset.data[0] = parseFloat(value);\n }\n }\n self.ctx.chart.update();\n}\n\nself.onResize = function() {\n self.ctx.chart.resize();\n}\n\nself.onDestroy = function() {\n self.ctx.chart.destroy();\n self.ctx.chart = null;\n}\n", |
|||
"settingsSchema": "", |
|||
"dataKeySettingsSchema": "{}\n", |
|||
"settingsDirective": "tb-chart-widget-settings", |
|||
"defaultConfig": "{\"datasources\":[{\"type\":\"function\",\"name\":\"function\",\"dataKeys\":[{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"First\",\"color\":\"#2196f3\",\"settings\":{},\"_hash\":0.15479322438769105,\"funcBody\":\"var value = (prevValue-50) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+50;\"},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Second\",\"color\":\"#4caf50\",\"settings\":{},\"_hash\":0.545701115289893,\"funcBody\":\"var value = (prevValue-20) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+20;\"},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Third\",\"color\":\"#f44336\",\"settings\":{},\"_hash\":0.2592906835158064,\"funcBody\":\"var value = (prevValue-40) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+40;\"},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Fourth\",\"color\":\"#ffc107\",\"settings\":{},\"_hash\":0.12880275585455747,\"funcBody\":\"var value = (prevValue-50) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+50;\"}]}],\"timewindow\":{\"realtime\":{\"timewindowMs\":60000}},\"showTitle\":true,\"backgroundColor\":\"#fff\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"8px\",\"settings\":{},\"title\":\"Bars\"}" |
|||
}, |
|||
"externalId": null, |
|||
"tags": [ |
|||
"bar", |
|||
"bar chart" |
|||
] |
|||
} |
|||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@ -0,0 +1,118 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.exception.ExceptionUtils; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.HousekeeperTaskProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.housekeeper.HousekeeperConfig; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; |
|||
|
|||
import javax.annotation.PreDestroy; |
|||
import java.util.LinkedHashSet; |
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
@TbCoreComponent |
|||
@Service |
|||
@Slf4j |
|||
public class HousekeeperReprocessingService { |
|||
|
|||
private final HousekeeperConfig config; |
|||
private final HousekeeperService housekeeperService; |
|||
private final QueueConsumerManager<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer; |
|||
private final TbQueueProducer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> producer; |
|||
private final TopicPartitionInfo submitTpi; |
|||
|
|||
private final ExecutorService consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-reprocessing-consumer")); |
|||
|
|||
public HousekeeperReprocessingService(HousekeeperConfig config, |
|||
@Lazy HousekeeperService housekeeperService, |
|||
TbCoreQueueFactory queueFactory) { |
|||
this.config = config; |
|||
this.housekeeperService = housekeeperService; |
|||
this.consumer = QueueConsumerManager.<TbProtoQueueMsg<ToHousekeeperServiceMsg>>builder() |
|||
.name("Housekeeper reprocessing") |
|||
.msgPackProcessor(this::processMsgs) |
|||
.pollInterval(config.getPollInterval()) |
|||
.consumerCreator(queueFactory::createHousekeeperReprocessingMsgConsumer) |
|||
.consumerExecutor(consumerExecutor) |
|||
.build(); |
|||
this.producer = queueFactory.createHousekeeperReprocessingMsgProducer(); |
|||
this.submitTpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build(); |
|||
} |
|||
|
|||
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
|||
public void afterStartUp() { |
|||
consumer.subscribe(); // Kafka topic for tasks reprocessing has only 1 partition, so only one TB Core will reprocess tasks
|
|||
consumer.launch(); |
|||
} |
|||
|
|||
private void processMsgs(List<TbProtoQueueMsg<ToHousekeeperServiceMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer) throws Exception { |
|||
Thread.sleep(config.getTaskReprocessingDelay()); |
|||
|
|||
for (TbProtoQueueMsg<ToHousekeeperServiceMsg> msg : msgs) { |
|||
log.trace("Reprocessing task: {}", msg); |
|||
try { |
|||
housekeeperService.processTask(msg.getValue()); |
|||
} catch (InterruptedException e) { |
|||
return; |
|||
} catch (Throwable e) { |
|||
log.error("Unexpected error during message reprocessing [{}]", msg, e); |
|||
submitForReprocessing(msg.getValue(), e); |
|||
} |
|||
} |
|||
consumer.commit(); |
|||
} |
|||
|
|||
public void submitForReprocessing(ToHousekeeperServiceMsg msg, Throwable error) { |
|||
HousekeeperTaskProto task = msg.getTask(); |
|||
Set<String> errors = new LinkedHashSet<>(task.getErrorsList()); |
|||
errors.add(StringUtils.truncate(ExceptionUtils.getStackTrace(error), 1024)); |
|||
msg = msg.toBuilder() |
|||
.setTask(task.toBuilder() |
|||
.setAttempt(task.getAttempt() + 1) |
|||
.clearErrors().addAllErrors(errors) |
|||
.build()) |
|||
.build(); |
|||
|
|||
log.trace("Submitting for reprocessing: {}", msg); |
|||
producer.send(submitTpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), null); |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void stop() { |
|||
consumer.stop(); |
|||
consumerExecutor.shutdownNow(); |
|||
log.info("Stopped Housekeeper reprocessing service"); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,166 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.TaskProcessingFailureTrigger; |
|||
import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.housekeeper.HousekeeperConfig; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.housekeeper.processor.HousekeeperTaskProcessor; |
|||
import org.thingsboard.server.service.housekeeper.stats.HousekeeperStatsService; |
|||
import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; |
|||
|
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.Future; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.TimeoutException; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@TbCoreComponent |
|||
@Service |
|||
@Slf4j |
|||
public class HousekeeperService { |
|||
|
|||
private final Map<HousekeeperTaskType, HousekeeperTaskProcessor<?>> taskProcessors; |
|||
|
|||
private final HousekeeperConfig config; |
|||
private final HousekeeperReprocessingService reprocessingService; |
|||
private final Optional<HousekeeperStatsService> statsService; |
|||
private final NotificationRuleProcessor notificationRuleProcessor; |
|||
private final QueueConsumerManager<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer; |
|||
|
|||
private final ExecutorService consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-consumer")); |
|||
private final ExecutorService taskExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-task-processor")); |
|||
|
|||
public HousekeeperService(HousekeeperConfig config, |
|||
HousekeeperReprocessingService reprocessingService, |
|||
TbCoreQueueFactory queueFactory, |
|||
Optional<HousekeeperStatsService> statsService, |
|||
NotificationRuleProcessor notificationRuleProcessor, |
|||
@Lazy List<HousekeeperTaskProcessor<?>> taskProcessors) { |
|||
this.config = config; |
|||
this.reprocessingService = reprocessingService; |
|||
this.statsService = statsService; |
|||
this.notificationRuleProcessor = notificationRuleProcessor; |
|||
this.consumer = QueueConsumerManager.<TbProtoQueueMsg<ToHousekeeperServiceMsg>>builder() |
|||
.name("Housekeeper") |
|||
.msgPackProcessor(this::processMsgs) |
|||
.pollInterval(config.getPollInterval()) |
|||
.consumerCreator(queueFactory::createHousekeeperMsgConsumer) |
|||
.consumerExecutor(consumerExecutor) |
|||
.build(); |
|||
this.taskProcessors = taskProcessors.stream().collect(Collectors.toMap(HousekeeperTaskProcessor::getTaskType, p -> p)); |
|||
} |
|||
|
|||
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
|||
public void afterStartUp() { |
|||
consumer.subscribe(); |
|||
consumer.launch(); |
|||
} |
|||
|
|||
private void processMsgs(List<TbProtoQueueMsg<ToHousekeeperServiceMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer) { |
|||
for (TbProtoQueueMsg<ToHousekeeperServiceMsg> msg : msgs) { |
|||
log.trace("Processing task: {}", msg); |
|||
try { |
|||
processTask(msg.getValue()); |
|||
} catch (InterruptedException e) { |
|||
return; |
|||
} catch (Throwable e) { |
|||
log.error("Unexpected error during message processing [{}]", msg, e); |
|||
reprocessingService.submitForReprocessing(msg.getValue(), e); |
|||
} |
|||
} |
|||
consumer.commit(); |
|||
} |
|||
|
|||
@SuppressWarnings("unchecked") |
|||
protected <T extends HousekeeperTask> void processTask(ToHousekeeperServiceMsg msg) throws Exception { |
|||
HousekeeperTask task = JacksonUtil.fromString(msg.getTask().getValue(), HousekeeperTask.class); |
|||
HousekeeperTaskType taskType = task.getTaskType(); |
|||
if (config.getDisabledTaskTypes().contains(taskType)) { |
|||
log.debug("Task type {} is disabled, ignoring {}", taskType, task); |
|||
return; |
|||
} |
|||
HousekeeperTaskProcessor<T> taskProcessor = (HousekeeperTaskProcessor<T>) taskProcessors.get(taskType); |
|||
if (taskProcessor == null) { |
|||
throw new IllegalArgumentException("Unsupported task type " + taskType); |
|||
} |
|||
|
|||
try { |
|||
long startTs = System.currentTimeMillis(); |
|||
Future<Object> future = taskExecutor.submit(() -> { |
|||
taskProcessor.process((T) task); |
|||
return null; |
|||
}); |
|||
future.get(config.getTaskProcessingTimeout(), TimeUnit.MILLISECONDS); |
|||
|
|||
long timing = System.currentTimeMillis() - startTs; |
|||
if (log.isDebugEnabled()) { |
|||
log.debug("[{}] Processed {} in {} ms (attempt {})", task.getTenantId(), task.getDescription(), timing, msg.getTask().getAttempt()); |
|||
} |
|||
statsService.ifPresent(statsService -> statsService.reportProcessed(taskType, msg, timing)); |
|||
} catch (InterruptedException e) { |
|||
throw e; |
|||
} catch (Throwable e) { |
|||
Throwable error = e; |
|||
if (e instanceof ExecutionException) { |
|||
error = e.getCause(); |
|||
} else if (e instanceof TimeoutException) { |
|||
error = new TimeoutException("Timeout after " + config.getTaskProcessingTimeout() + " seconds"); |
|||
} |
|||
|
|||
if (msg.getTask().getAttempt() < config.getMaxReprocessingAttempts()) { |
|||
log.warn("[{}] Failed to process {} (attempt {}), submitting for reprocessing", |
|||
task.getTenantId(), task.getDescription(), msg.getTask().getAttempt(), error); |
|||
reprocessingService.submitForReprocessing(msg, error); |
|||
} else { |
|||
log.error("[{}] Failed to process task in {} attempts: {}", task.getTenantId(), msg.getTask().getAttempt(), msg, e); |
|||
notificationRuleProcessor.process(TaskProcessingFailureTrigger.builder() |
|||
.task(task) |
|||
.error(error) |
|||
.attempt(msg.getTask().getAttempt()) |
|||
.build()); |
|||
} |
|||
statsService.ifPresent(statsService -> statsService.reportFailure(taskType, msg)); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void stop() { |
|||
consumer.stop(); |
|||
consumerExecutor.shutdownNow(); |
|||
log.info("Stopped Housekeeper service"); |
|||
} |
|||
|
|||
} |
|||
@ -1,103 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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; |
|||
|
|||
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.ListeningExecutorService; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.springframework.transaction.event.TransactionalEventListener; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent; |
|||
import org.thingsboard.server.dao.housekeeper.HouseKeeperService; |
|||
import org.thingsboard.server.service.entitiy.alarm.TbAlarmService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class InMemoryHouseKeeperServiceService implements HouseKeeperService { |
|||
|
|||
final TbAlarmService alarmService; |
|||
|
|||
ListeningExecutorService executor; |
|||
|
|||
AtomicInteger queueSize = new AtomicInteger(); |
|||
AtomicInteger totalProcessedCounter = new AtomicInteger(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
log.debug("Starting HouseKeeper service"); |
|||
executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper"))); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
if (executor != null) { |
|||
log.debug("Stopping HouseKeeper service"); |
|||
executor.shutdown(); |
|||
} |
|||
} |
|||
|
|||
@TransactionalEventListener(fallbackExecution = true) |
|||
public void handleEvent(DeleteEntityEvent<?> event) { |
|||
log.trace("[{}] DeleteEntityEvent handler: {}", event.getTenantId(), event); |
|||
EntityId entityId = event.getEntityId(); |
|||
if (EntityType.USER.equals(entityId.getEntityType())) { |
|||
unassignDeletedUserAlarms(event.getTenantId(), (User) event.getEntity(), event.getTs()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<List<AlarmId>> unassignDeletedUserAlarms(TenantId tenantId, User user, long unassignTs) { |
|||
log.debug("[{}][{}] unassignDeletedUserAlarms submitting, pending queue size: {} ", tenantId, user.getId().getId(), queueSize.get()); |
|||
queueSize.incrementAndGet(); |
|||
ListenableFuture<List<AlarmId>> future = executor.submit(() -> alarmService.unassignDeletedUserAlarms(tenantId, user, unassignTs)); |
|||
Futures.addCallback(future, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(List<AlarmId> alarmIds) { |
|||
queueSize.decrementAndGet(); |
|||
totalProcessedCounter.incrementAndGet(); |
|||
log.debug("[{}][{}] unassignDeletedUserAlarms finished, pending queue size: {}, total processed count: {} ", |
|||
tenantId, user.getId().getId(), queueSize.get(), totalProcessedCounter.get()); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable throwable) { |
|||
queueSize.decrementAndGet(); |
|||
totalProcessedCounter.incrementAndGet(); |
|||
log.error("[{}][{}] unassignDeletedUserAlarms failed, pending queue size: {}, total processed count: {}", |
|||
tenantId, user.getId().getId(), queueSize.get(), totalProcessedCounter.get(), throwable); |
|||
} |
|||
}, MoreExecutors.directExecutor()); |
|||
return future; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,80 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.EntityType; |
|||
import org.thingsboard.server.common.data.housekeeper.AlarmsDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.util.TbPair; |
|||
import org.thingsboard.server.dao.alarm.AlarmService; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class AlarmsDeletionTaskProcessor extends HousekeeperTaskProcessor<AlarmsDeletionHousekeeperTask> { |
|||
|
|||
private final AlarmService alarmService; |
|||
|
|||
@Override |
|||
public void process(AlarmsDeletionHousekeeperTask task) throws Exception { |
|||
EntityId entityId = task.getEntityId(); |
|||
EntityType entityType = entityId.getEntityType(); |
|||
TenantId tenantId = task.getTenantId(); |
|||
|
|||
if (entityType == EntityType.DEVICE || entityType == EntityType.ASSET) { |
|||
if (task.getAlarms() == null) { |
|||
AlarmId lastId = null; |
|||
long lastCreatedTime = 0; |
|||
while (true) { |
|||
List<TbPair<UUID, Long>> alarms = alarmService.findAlarmIdsByOriginatorId(tenantId, entityId, lastCreatedTime, lastId, 128); |
|||
if (alarms.isEmpty()) { |
|||
break; |
|||
} |
|||
|
|||
housekeeperClient.submitTask(new AlarmsDeletionHousekeeperTask(tenantId, entityId, alarms.stream().map(TbPair::getFirst).toList())); |
|||
|
|||
TbPair<UUID, Long> last = alarms.get(alarms.size() - 1); |
|||
lastId = new AlarmId(last.getFirst()); |
|||
lastCreatedTime = last.getSecond(); |
|||
log.debug("[{}][{}][{}] Submitted task for deleting {} alarms", tenantId, entityType, entityId, alarms.size()); |
|||
} |
|||
} else { |
|||
for (UUID alarmId : task.getAlarms()) { |
|||
alarmService.delAlarm(tenantId, new AlarmId(alarmId)); |
|||
} |
|||
log.debug("[{}][{}][{}] Deleted {} alarms", tenantId, entityType, entityId, task.getAlarms().size()); |
|||
} |
|||
} |
|||
|
|||
int count = alarmService.deleteEntityAlarmRecords(tenantId, entityId); |
|||
log.debug("[{}][{}][{}] Deleted {} entity alarms", tenantId, entityType, entityId, count); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_ALARMS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.AlarmsUnassignHousekeeperTask; |
|||
import org.thingsboard.server.service.entitiy.alarm.TbAlarmService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class AlarmsUnassignTaskProcessor extends HousekeeperTaskProcessor<AlarmsUnassignHousekeeperTask> { |
|||
|
|||
private final TbAlarmService alarmService; |
|||
|
|||
@Override |
|||
public void process(AlarmsUnassignHousekeeperTask task) throws Exception { |
|||
List<AlarmId> alarms = alarmService.unassignDeletedUserAlarms(task.getTenantId(), (UserId) task.getEntityId(), task.getUserTitle(), task.getTs()); |
|||
log.debug("[{}][{}] Unassigned {} alarms", task.getTenantId(), task.getEntityId(), alarms.size()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.UNASSIGN_ALARMS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.dao.attributes.AttributesService; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class AttributesDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final AttributesService attributesService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
int deletedCount = attributesService.removeAllByEntityId(task.getTenantId(), task.getEntityId()); |
|||
log.debug("[{}][{}][{}] Deleted {} attributes", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), deletedCount); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_ATTRIBUTES; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,57 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.EntityType; |
|||
import org.thingsboard.server.common.data.housekeeper.EntitiesDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.entity.EntityDaoService; |
|||
import org.thingsboard.server.dao.entity.EntityServiceRegistry; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class EntitiesDeletionTaskProcessor extends HousekeeperTaskProcessor<EntitiesDeletionHousekeeperTask> { |
|||
|
|||
private final EntityServiceRegistry entityServiceRegistry; |
|||
|
|||
@Override |
|||
public void process(EntitiesDeletionHousekeeperTask task) throws Exception { |
|||
EntityType entityType = task.getEntityType(); |
|||
TenantId tenantId = task.getTenantId(); |
|||
EntityDaoService entityService = entityServiceRegistry.getServiceByEntityType(entityType); |
|||
|
|||
for (UUID entityUuid : task.getEntities()) { |
|||
EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityType, entityUuid); |
|||
entityService.deleteEntity(tenantId, entityId, true); |
|||
} |
|||
log.debug("[{}] Deleted {} {}s", tenantId, task.getEntities().size(), entityType.getNormalName().toLowerCase()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_ENTITIES; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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 org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.dao.event.EventService; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class EventsDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final EventService eventService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
eventService.removeEvents(task.getTenantId(), task.getEntityId(), null, 0L, System.currentTimeMillis()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_EVENTS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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 org.springframework.beans.factory.annotation.Autowired; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.msg.housekeeper.HousekeeperClient; |
|||
|
|||
public abstract class HousekeeperTaskProcessor<T extends HousekeeperTask> { |
|||
|
|||
@Autowired |
|||
protected HousekeeperClient housekeeperClient; |
|||
|
|||
public abstract void process(T task) throws Exception; |
|||
|
|||
public abstract HousekeeperTaskType getTaskType(); |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.LatestTsDeletionHousekeeperTask; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class LatestTsDeletionTaskProcessor extends HousekeeperTaskProcessor<LatestTsDeletionHousekeeperTask> { |
|||
|
|||
private final TimeseriesService timeseriesService; |
|||
|
|||
@Override |
|||
public void process(LatestTsDeletionHousekeeperTask task) throws Exception { |
|||
timeseriesService.removeLatest(task.getTenantId(), task.getEntityId(), List.of(task.getKey())).get(); |
|||
log.debug("[{}][{}][{}] Deleted latest telemetry for key '{}'", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getKey()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_LATEST_TS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data.housekeeper.LatestTsDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.TsHistoryDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class TelemetryDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final TimeseriesService timeseriesService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
TenantId tenantId = task.getTenantId(); |
|||
EntityId entityId = task.getEntityId(); |
|||
List<String> keys = timeseriesService.findAllKeysByEntityIds(tenantId, List.of(entityId)); |
|||
|
|||
for (String key : keys) { |
|||
var latestTsDeletionTask = new LatestTsDeletionHousekeeperTask(tenantId, entityId, key); |
|||
housekeeperClient.submitTask(latestTsDeletionTask); |
|||
|
|||
var tsHistoryDeletionTask = new TsHistoryDeletionHousekeeperTask(tenantId, entityId, key); |
|||
housekeeperClient.submitTask(tsHistoryDeletionTask); |
|||
} |
|||
|
|||
log.trace("[{}][{}][{}] Submitted latest and ts history deletion tasks for {} keys", tenantId, entityId.getEntityType(), entityId, keys.size()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_TELEMETRY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,63 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.EntityType; |
|||
import org.thingsboard.server.common.data.housekeeper.EntitiesDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.TenantEntitiesDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.Dao; |
|||
import org.thingsboard.server.dao.entity.EntityDaoRegistry; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class TenantEntitiesDeletionTaskProcessor extends HousekeeperTaskProcessor<TenantEntitiesDeletionHousekeeperTask> { |
|||
|
|||
private final EntityDaoRegistry entityDaoRegistry; |
|||
|
|||
@Override |
|||
public void process(TenantEntitiesDeletionHousekeeperTask task) throws Exception { |
|||
EntityType entityType = task.getEntityType(); |
|||
TenantId tenantId = task.getTenantId(); |
|||
Dao<?> entityDao = entityDaoRegistry.getDao(entityType); |
|||
|
|||
UUID last = null; |
|||
while (true) { |
|||
List<UUID> entities = entityDao.findIdsByTenantIdAndIdOffset(tenantId, last, 128); |
|||
if (entities.isEmpty()) { |
|||
break; |
|||
} |
|||
|
|||
housekeeperClient.submitTask(new EntitiesDeletionHousekeeperTask(tenantId, entityType, entities)); |
|||
last = entities.get(entities.size() - 1); |
|||
log.debug("[{}] Submitted task for deleting {} {}s", tenantId, entities.size(), entityType.getNormalName().toLowerCase()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_TENANT_ENTITIES; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.TsHistoryDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.kv.BaseDeleteTsKvQuery; |
|||
import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class TsHistoryDeletionTaskProcessor extends HousekeeperTaskProcessor<TsHistoryDeletionHousekeeperTask> { |
|||
|
|||
private final TimeseriesService timeseriesService; |
|||
|
|||
@Override |
|||
public void process(TsHistoryDeletionHousekeeperTask task) throws Exception { |
|||
DeleteTsKvQuery deleteQuery = new BaseDeleteTsKvQuery(task.getKey(), 0, System.currentTimeMillis(), false, false); |
|||
timeseriesService.remove(task.getTenantId(), task.getEntityId(), List.of(deleteQuery)).get(); |
|||
log.debug("[{}][{}][{}] Deleted timeseries history for key '{}'", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getKey()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_TS_HISTORY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,125 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.stats; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.stats.DefaultCounter; |
|||
import org.thingsboard.server.common.stats.StatsCounter; |
|||
import org.thingsboard.server.common.stats.StatsFactory; |
|||
import org.thingsboard.server.common.stats.StatsTimer; |
|||
import org.thingsboard.server.common.stats.StatsType; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.EnumMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Objects; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@ConditionalOnProperty(name = "queue.core.housekeeper.stats.enabled", havingValue = "true", matchIfMissing = true) |
|||
public class HousekeeperStatsService { |
|||
|
|||
private final Map<HousekeeperTaskType, HousekeeperStats> stats = new EnumMap<>(HousekeeperTaskType.class); |
|||
|
|||
public HousekeeperStatsService(StatsFactory statsFactory) { |
|||
for (HousekeeperTaskType taskType : HousekeeperTaskType.values()) { |
|||
stats.put(taskType, new HousekeeperStats(taskType, statsFactory)); |
|||
} |
|||
} |
|||
|
|||
@Scheduled(initialDelayString = "${queue.core.housekeeper.stats.print-interval-ms:60000}", |
|||
fixedDelayString = "${queue.core.housekeeper.stats.print-interval-ms:60000}") |
|||
private void reportStats() { |
|||
String statsStr = stats.values().stream().map(stats -> { |
|||
String countersStr = stats.getCounters().stream() |
|||
.filter(counter -> counter.get() > 0) |
|||
.map(counter -> counter.getName() + " = [" + counter.get() + "]") |
|||
.collect(Collectors.joining(" ")); |
|||
if (countersStr.isEmpty()) { |
|||
return null; |
|||
} else { |
|||
return stats.getTaskType() + " " + countersStr + " avgProcessingTime [" + stats.getProcessingTimer().getAvg() + " ms]"; |
|||
} |
|||
}).filter(Objects::nonNull).collect(Collectors.joining(", ")); |
|||
|
|||
if (!statsStr.isEmpty()) { |
|||
stats.values().forEach(HousekeeperStats::reset); |
|||
log.info("Housekeeper stats: {}", statsStr); |
|||
} |
|||
} |
|||
|
|||
public void reportProcessed(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg, long timing) { |
|||
HousekeeperStats stats = this.stats.get(taskType); |
|||
if (msg.getTask().getErrorsCount() == 0) { |
|||
stats.getProcessedCounter().increment(); |
|||
} else { |
|||
stats.getReprocessedCounter().increment(); |
|||
} |
|||
stats.getProcessingTimer().record(timing); |
|||
} |
|||
|
|||
public void reportFailure(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg) { |
|||
HousekeeperStats stats = this.stats.get(taskType); |
|||
if (msg.getTask().getErrorsCount() == 0) { |
|||
stats.getFailedProcessingCounter().increment(); |
|||
} else { |
|||
stats.getFailedReprocessingCounter().increment(); |
|||
} |
|||
} |
|||
|
|||
@Getter |
|||
static class HousekeeperStats { |
|||
private final HousekeeperTaskType taskType; |
|||
private final List<StatsCounter> counters = new ArrayList<>(); |
|||
|
|||
private final StatsCounter processedCounter; |
|||
private final StatsCounter failedProcessingCounter; |
|||
private final StatsCounter reprocessedCounter; |
|||
private final StatsCounter failedReprocessingCounter; |
|||
|
|||
private final StatsTimer processingTimer; |
|||
|
|||
public HousekeeperStats(HousekeeperTaskType taskType, StatsFactory statsFactory) { |
|||
this.taskType = taskType; |
|||
this.processedCounter = register("processed", statsFactory); |
|||
this.failedProcessingCounter = register("failedProcessing", statsFactory); |
|||
this.reprocessedCounter = register("reprocessed", statsFactory); |
|||
this.failedReprocessingCounter = register("failedReprocessing", statsFactory); |
|||
this.processingTimer = statsFactory.createTimer(StatsType.HOUSEKEEPER, "processingTime", "taskType", taskType.name()); |
|||
} |
|||
|
|||
private StatsCounter register(String statsName, StatsFactory statsFactory) { |
|||
StatsCounter counter = statsFactory.createStatsCounter(StatsType.HOUSEKEEPER.getName(), statsName, "taskType", taskType.name()); |
|||
counters.add(counter); |
|||
return counter; |
|||
} |
|||
|
|||
public void reset() { |
|||
counters.forEach(DefaultCounter::clear); |
|||
processingTimer.reset(); |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,53 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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 org.apache.commons.lang3.exception.ExceptionUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.notification.info.TaskProcessingFailureNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.TaskProcessingFailureTrigger; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.TaskProcessingFailureNotificationRuleTriggerConfig; |
|||
|
|||
@Service |
|||
public class TaskProcessingFailureTriggerProcessor implements NotificationRuleTriggerProcessor<TaskProcessingFailureTrigger, TaskProcessingFailureNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(TaskProcessingFailureTrigger trigger, TaskProcessingFailureNotificationRuleTriggerConfig triggerConfig) { |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public TaskProcessingFailureNotificationInfo constructNotificationInfo(TaskProcessingFailureTrigger trigger) { |
|||
HousekeeperTask task = trigger.getTask(); |
|||
return TaskProcessingFailureNotificationInfo.builder() |
|||
.tenantId(task.getTenantId()) |
|||
.entityId(task.getEntityId()) |
|||
.taskType(task.getTaskType()) |
|||
.taskDescription(task.getDescription()) |
|||
.error(StringUtils.truncate(ExceptionUtils.getStackTrace(trigger.getError()), 1024)) |
|||
.attempt(trigger.getAttempt()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.TASK_PROCESSING_FAILURE; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,109 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.queue.consumer; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueMsg; |
|||
|
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.function.Supplier; |
|||
|
|||
@Slf4j |
|||
public class QueueConsumerManager<M extends TbQueueMsg> { |
|||
|
|||
private final String name; |
|||
private final MsgPackProcessor<M> msgPackProcessor; |
|||
private final long pollInterval; |
|||
private final ExecutorService consumerExecutor; |
|||
private final String threadPrefix; |
|||
|
|||
@Getter |
|||
private final TbQueueConsumer<M> consumer; |
|||
private volatile boolean stopped; |
|||
|
|||
@Builder |
|||
public QueueConsumerManager(String name, MsgPackProcessor<M> msgPackProcessor, |
|||
long pollInterval, Supplier<TbQueueConsumer<M>> consumerCreator, |
|||
ExecutorService consumerExecutor, String threadPrefix) { |
|||
this.name = name; |
|||
this.pollInterval = pollInterval; |
|||
this.msgPackProcessor = msgPackProcessor; |
|||
this.consumerExecutor = consumerExecutor; |
|||
this.threadPrefix = threadPrefix; |
|||
this.consumer = consumerCreator.get(); |
|||
} |
|||
|
|||
public void subscribe() { |
|||
consumer.subscribe(); |
|||
} |
|||
|
|||
public void subscribe(Set<TopicPartitionInfo> partitions) { |
|||
consumer.subscribe(partitions); |
|||
} |
|||
|
|||
public void launch() { |
|||
log.info("[{}] Launching consumer", name); |
|||
consumerExecutor.submit(() -> { |
|||
if (threadPrefix != null) { |
|||
ThingsBoardThreadFactory.addThreadNamePrefix(threadPrefix); |
|||
} |
|||
try { |
|||
consumerLoop(consumer); |
|||
} catch (Throwable e) { |
|||
log.error("Failure in consumer loop", e); |
|||
} |
|||
log.info("[{}] Consumer stopped", name); |
|||
}); |
|||
} |
|||
|
|||
private void consumerLoop(TbQueueConsumer<M> consumer) { |
|||
while (!stopped && !consumer.isStopped()) { |
|||
try { |
|||
List<M> msgs = consumer.poll(pollInterval); |
|||
if (msgs.isEmpty()) { |
|||
continue; |
|||
} |
|||
msgPackProcessor.process(msgs, consumer); |
|||
} catch (Exception e) { |
|||
if (!consumer.isStopped()) { |
|||
log.warn("Failed to process messages from queue", e); |
|||
try { |
|||
Thread.sleep(pollInterval); |
|||
} catch (InterruptedException interruptedException) { |
|||
log.trace("Failed to wait until the server has capacity to handle new requests", interruptedException); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
public void stop() { |
|||
log.debug("[{}] Stopping consumer", name); |
|||
stopped = true; |
|||
consumer.unsubscribe(); |
|||
} |
|||
|
|||
public interface MsgPackProcessor<M extends TbQueueMsg> { |
|||
void process(List<M> msgs, TbQueueConsumer<M> consumer) throws Exception; |
|||
} |
|||
} |
|||
@ -0,0 +1,478 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import com.fasterxml.jackson.databind.node.TextNode; |
|||
import org.junit.After; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.mockito.ArgumentMatcher; |
|||
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.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.rule.engine.metadata.TbGetAttributesNode; |
|||
import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration; |
|||
import org.thingsboard.server.common.data.ApiUsageState; |
|||
import org.thingsboard.server.common.data.AttributeScope; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.EventInfo; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.alarm.AlarmSeverity; |
|||
import org.thingsboard.server.common.data.alarm.EntityAlarm; |
|||
import org.thingsboard.server.common.data.asset.Asset; |
|||
import org.thingsboard.server.common.data.event.EventType; |
|||
import org.thingsboard.server.common.data.event.LifecycleEvent; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.AssetId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; |
|||
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
|||
import org.thingsboard.server.common.data.kv.StringDataEntry; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
import org.thingsboard.server.common.data.msg.TbNodeConnectionType; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
import org.thingsboard.server.common.data.relation.EntityRelation; |
|||
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
|||
import org.thingsboard.server.common.data.rule.RuleChain; |
|||
import org.thingsboard.server.common.data.rule.RuleChainMetaData; |
|||
import org.thingsboard.server.common.data.rule.RuleChainType; |
|||
import org.thingsboard.server.common.data.rule.RuleNode; |
|||
import org.thingsboard.server.controller.AbstractControllerTest; |
|||
import org.thingsboard.server.dao.alarm.AlarmDao; |
|||
import org.thingsboard.server.dao.alarm.AlarmService; |
|||
import org.thingsboard.server.dao.attributes.AttributesService; |
|||
import org.thingsboard.server.dao.entity.EntityServiceRegistry; |
|||
import org.thingsboard.server.dao.event.EventService; |
|||
import org.thingsboard.server.dao.relation.RelationService; |
|||
import org.thingsboard.server.dao.rule.RuleChainService; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
import org.thingsboard.server.dao.usagerecord.ApiUsageStateDao; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.HousekeeperTaskProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
import org.thingsboard.server.service.housekeeper.processor.TsHistoryDeletionTaskProcessor; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Arrays; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.TimeoutException; |
|||
import java.util.function.Function; |
|||
import java.util.function.Predicate; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.argThat; |
|||
import static org.mockito.Mockito.doCallRealMethod; |
|||
import static org.mockito.Mockito.doThrow; |
|||
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 = { |
|||
"transport.http.enabled=true", |
|||
"queue.core.housekeeper.task-reprocessing-delay-ms=2000", |
|||
"queue.core.housekeeper.poll-interval-ms=1000", |
|||
"queue.core.housekeeper.max-reprocessing-attempts=5" |
|||
}) |
|||
public class HousekeeperServiceTest extends AbstractControllerTest { |
|||
|
|||
@SpyBean |
|||
private HousekeeperService housekeeperService; |
|||
@SpyBean |
|||
private HousekeeperReprocessingService housekeeperReprocessingService; |
|||
@Autowired |
|||
private EventService eventService; |
|||
@Autowired |
|||
private TimeseriesService timeseriesService; |
|||
@Autowired |
|||
private AttributesService attributesService; |
|||
@Autowired |
|||
private RuleChainService ruleChainService; |
|||
@Autowired |
|||
private AlarmService alarmService; |
|||
@Autowired |
|||
private AlarmDao alarmDao; |
|||
@Autowired |
|||
private RelationService relationService; |
|||
@Autowired |
|||
private ApiUsageStateDao apiUsageStateDao; |
|||
@Autowired |
|||
private EntityServiceRegistry entityServiceRegistry; |
|||
@SpyBean |
|||
private TsHistoryDeletionTaskProcessor tsHistoryDeletionTaskProcessor; |
|||
|
|||
private TenantId tenantId; |
|||
|
|||
private static final String TELEMETRY_KEY = "telemetry1"; |
|||
private static final String ATTRIBUTE_KEY = "_attribute1"; |
|||
private static final String KV_VALUE = "ewfewfwef"; |
|||
|
|||
@Before |
|||
public void setUp() throws Exception { |
|||
loginTenantAdmin(); |
|||
this.tenantId = super.tenantId; |
|||
} |
|||
|
|||
@After |
|||
public void tearDown() throws Exception { |
|||
} |
|||
|
|||
@Test |
|||
public void whenDeviceIsDeleted_thenCleanUpRelatedData() throws Exception { |
|||
Device device = createDevice("test", "test"); |
|||
createRelatedData(device.getId()); |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoRelatedData(device.getId()); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenRuleChainIsDeleted_thenCleanUpRelatedData() throws Exception { |
|||
RuleChainMetaData ruleChainMetaData = createRuleChain(); |
|||
RuleChainId ruleChainId = ruleChainMetaData.getRuleChainId(); |
|||
RuleNodeId ruleNode1Id = ruleChainMetaData.getNodes().get(0).getId(); |
|||
RuleNodeId ruleNode2Id = ruleChainMetaData.getNodes().get(1).getId(); |
|||
createRelatedData(ruleChainId); |
|||
createRelatedData(ruleNode1Id); |
|||
createRelatedData(ruleNode2Id); |
|||
|
|||
doDelete("/api/ruleChain/" + ruleChainId).andExpect(status().isOk()); |
|||
|
|||
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoRelatedData(ruleNode1Id); |
|||
verifyNoRelatedData(ruleNode2Id); |
|||
verifyNoRelatedData(ruleChainId); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenUserIsDeleted_thenCleanUpRelatedData() throws Exception { |
|||
Device device = createDevice("test", "test"); |
|||
UserId userId = customerUserId; |
|||
createRelatedData(userId); |
|||
Alarm alarm = Alarm.builder() |
|||
.type("test") |
|||
.tenantId(tenantId) |
|||
.originator(device.getId()) |
|||
.severity(AlarmSeverity.MAJOR) |
|||
.build(); |
|||
alarm = doPost("/api/alarm", alarm, Alarm.class); |
|||
AlarmId alarmId = alarm.getId(); |
|||
alarm = doPost("/api/alarm/" + alarmId + "/assign/" + userId, "", Alarm.class); |
|||
assertThat(alarm.getAssigneeId()).isEqualTo(userId); |
|||
assertThat(alarmService.findAlarmIdsByAssigneeId(tenantId, userId, new PageLink(100)).getData()).isNotEmpty(); |
|||
|
|||
doDelete("/api/user/" + userId).andExpect(status().isOk()); |
|||
|
|||
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoRelatedData(userId); |
|||
assertThat(alarmService.findAlarmById(tenantId, alarmId).getAssigneeId()).isNull(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenDeviceIsDeleted_thenDeleteAllAlarms() throws Exception { |
|||
Device device = createDevice("test", "test"); |
|||
for (int i = 1; i <= 1000; i++) { |
|||
createAlarm(device.getId()); |
|||
} |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoAlarms(device.getId()); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenTenantIsDeleted_thenDeleteAllEntitiesAndCleanUpRelatedData() throws Exception { |
|||
loginDifferentTenant(); |
|||
tenantId = differentTenantId; |
|||
|
|||
createRelatedData(tenantId); |
|||
createDifferentTenantCustomer(); |
|||
createRelatedData(differentTenantCustomerId); |
|||
loginDifferentTenant(); |
|||
|
|||
List<DeviceId> devices = new ArrayList<>(); |
|||
for (int i = 1; i <= 300; i++) { |
|||
Device device = createDevice("test" + i, "test" + i); |
|||
devices.add(device.getId()); |
|||
} |
|||
DeviceId firstDevice = devices.get(0); |
|||
createRelatedData(firstDevice); |
|||
DeviceId lastDevice = devices.get(devices.size() - 1); |
|||
createRelatedData(lastDevice); |
|||
|
|||
Asset asset = createAsset(); |
|||
createRelatedData(asset.getId()); |
|||
createRelation(firstDevice, asset.getId()); |
|||
createAlarm(firstDevice, asset.getId()); |
|||
|
|||
RuleChainMetaData ruleChainMetaData = createRuleChain(); |
|||
RuleChainId ruleChainId = ruleChainMetaData.getRuleChainId(); |
|||
RuleNodeId ruleNode1Id = ruleChainMetaData.getNodes().get(0).getId(); |
|||
RuleNodeId ruleNode2Id = ruleChainMetaData.getNodes().get(1).getId(); |
|||
createRelatedData(ruleChainId); |
|||
createRelatedData(ruleNode1Id); |
|||
createRelatedData(ruleNode2Id); |
|||
|
|||
UserId userId = savedDifferentTenantUser.getId(); |
|||
createRelatedData(userId); |
|||
|
|||
ApiUsageState tenantApiUsageState = apiUsageStateDao.findApiUsageStateByEntityId(differentTenantId); |
|||
|
|||
loginSysAdmin(); |
|||
deleteDifferentTenant(); |
|||
|
|||
await().atMost(60, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
for (DeviceId deviceId : devices) { |
|||
verifyNoRelatedData(deviceId); |
|||
} |
|||
verifyNoRelatedData(asset.getId()); |
|||
verifyNoRelatedData(ruleNode1Id); |
|||
verifyNoRelatedData(ruleNode2Id); |
|||
verifyNoRelatedData(ruleChainId); |
|||
verifyNoRelatedData(userId); |
|||
verifyNoRelatedData(differentTenantCustomerId); |
|||
verifyNoRelatedData(tenantApiUsageState.getId()); |
|||
verifyNoRelatedData(tenantId); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenTaskProcessingFails_thenReprocess() throws Exception { |
|||
TimeoutException error = new TimeoutException("Test timeout"); |
|||
doThrow(error).when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
|
|||
Device device = createDevice("test", "test"); |
|||
createRelatedData(device.getId()); |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
int attempts = 2; |
|||
await().atMost(30, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
for (int i = 0; i <= attempts; i++) { |
|||
int attempt = i; |
|||
verify(housekeeperReprocessingService).submitForReprocessing(argThat(getTaskMatcher(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, |
|||
task -> task.getAttempt() == attempt)), argThat(e -> e.getMessage().equals(error.getMessage()))); |
|||
} |
|||
}); |
|||
|
|||
assertThat(getTimeseriesHistory(device.getId())).isNotEmpty(); |
|||
doCallRealMethod().when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
assertThat(getTimeseriesHistory(device.getId())).isEmpty(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenReprocessingAttemptsExceeded_thenDropTheTask() throws Exception { |
|||
TimeoutException error = new TimeoutException("Test timeout"); |
|||
doThrow(error).when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
|
|||
Device device = createDevice("test", "test"); |
|||
createRelatedData(device.getId()); |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
int maxAttempts = 5; |
|||
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
for (int i = 1; i <= maxAttempts; i++) { |
|||
verifyTaskProcessing(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, i); |
|||
} |
|||
}); |
|||
|
|||
Mockito.clearInvocations(housekeeperService); |
|||
doCallRealMethod().when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
TimeUnit.SECONDS.sleep(2); |
|||
verify(housekeeperService, never()).processTask(argThat(getTaskMatcher(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, null))); |
|||
} |
|||
|
|||
private void verifyTaskProcessing(EntityId entityId, HousekeeperTaskType taskType, int expectedAttempt) throws Exception { |
|||
verify(housekeeperService).processTask(argThat(getTaskMatcher(entityId, taskType, task -> task.getAttempt() == expectedAttempt))); |
|||
} |
|||
|
|||
private ArgumentMatcher<ToHousekeeperServiceMsg> getTaskMatcher(EntityId entityId, HousekeeperTaskType taskType, |
|||
Predicate<HousekeeperTaskProto> additionalCheck) { |
|||
return msg -> { |
|||
HousekeeperTask task = JacksonUtil.fromString(msg.getTask().getValue(), HousekeeperTask.class); |
|||
return task.getEntityId().equals(entityId) && task.getTaskType() == taskType && (additionalCheck == null || additionalCheck.test(msg.getTask())); |
|||
}; |
|||
} |
|||
|
|||
private void createRelatedData(EntityId entityId) throws Exception { |
|||
createTelemetry(entityId); |
|||
for (AttributeScope scope : AttributeScope.values()) { |
|||
createAttribute(entityId, scope, scope + ATTRIBUTE_KEY); |
|||
} |
|||
createEvent(entityId); |
|||
} |
|||
|
|||
private void verifyNoRelatedData(EntityId entityId) throws Exception { |
|||
assertThat(entityServiceRegistry.getServiceByEntityType(entityId.getEntityType()).findEntity(tenantId, entityId)).isEmpty(); |
|||
|
|||
assertThat(getLatestTelemetry(entityId)).isNull(); |
|||
assertThat(getTimeseriesHistory(entityId)).isEmpty(); |
|||
for (AttributeScope scope : AttributeScope.values()) { |
|||
assertThat(attributesService.findAll(tenantId, entityId, scope).get()).isEmpty(); |
|||
} |
|||
assertThat(getEvents(entityId)).isEmpty(); |
|||
assertThat(alarmDao.findEntityAlarmRecordsByEntityId(tenantId, entityId)).isEmpty(); |
|||
verifyNoAlarms(entityId); |
|||
assertThat(relationService.findByTo(tenantId, entityId, RelationTypeGroup.COMMON)).isEmpty(); |
|||
assertThat(relationService.findByFrom(tenantId, entityId, RelationTypeGroup.COMMON)).isEmpty(); |
|||
} |
|||
|
|||
private void verifyNoAlarms(EntityId entityId) { |
|||
assertThat(alarmService.findAlarmIdsByOriginatorId(tenantId, entityId, 0, null, 10)).isEmpty(); |
|||
} |
|||
|
|||
private void createAttribute(EntityId entityId, AttributeScope scope, String key) throws Exception { |
|||
attributesService.save(tenantId, entityId, scope, new BaseAttributeKvEntry(System.currentTimeMillis(), new StringDataEntry(key, KV_VALUE))).get(); |
|||
} |
|||
|
|||
private void createTelemetry(EntityId entityId) throws Exception { |
|||
timeseriesService.save(tenantId, entityId, new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry(TELEMETRY_KEY, KV_VALUE))).get(); |
|||
} |
|||
|
|||
private void createEvent(EntityId entityId) { |
|||
LifecycleEvent event = LifecycleEvent.builder() |
|||
.tenantId(tenantId) |
|||
.entityId(entityId.getId()) |
|||
.serviceId("test") |
|||
.lcEventType("test") |
|||
.success(true) |
|||
.build(); |
|||
eventService.saveAsync(event); |
|||
await().atMost(10, TimeUnit.SECONDS) |
|||
.until(() -> !getEvents(entityId).isEmpty()); |
|||
|
|||
} |
|||
|
|||
private void createRelation(DeviceId to, AssetId from) { |
|||
EntityRelation relation = new EntityRelation(from, to, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON); |
|||
relationService.saveRelation(tenantId, relation); |
|||
} |
|||
|
|||
private void createAlarm(DeviceId deviceId, EntityId propagatedEntityId) { |
|||
Alarm alarm = doPost("/api/alarm", Alarm.builder() |
|||
.tenantId(tenantId) |
|||
.originator(deviceId) |
|||
.severity(AlarmSeverity.CRITICAL) |
|||
.type("test alarm for " + deviceId) |
|||
.propagate(true) |
|||
.build(), Alarm.class); |
|||
|
|||
List<EntityAlarm> entityAlarms = alarmDao.findEntityAlarmRecords(tenantId, alarm.getId()); |
|||
assertThat(entityAlarms).anyMatch(entityAlarm -> entityAlarm.getEntityId().equals(deviceId) && entityAlarm.getAlarmType().equals(alarm.getType())); |
|||
assertThat(entityAlarms).anyMatch(entityAlarm -> entityAlarm.getEntityId().equals(propagatedEntityId) && entityAlarm.getAlarmType().equals(alarm.getType())); |
|||
assertThat(alarmService.findAlarmIdsByOriginatorId(tenantId, deviceId, 0, null, 10)).isNotEmpty(); |
|||
} |
|||
|
|||
private void createAlarm(DeviceId deviceId) { |
|||
Alarm alarm = doPost("/api/alarm", Alarm.builder() |
|||
.tenantId(tenantId) |
|||
.originator(deviceId) |
|||
.severity(AlarmSeverity.CRITICAL) |
|||
.type("test alarm for " + deviceId + " " + RandomStringUtils.randomAlphabetic(10)) |
|||
.build(), Alarm.class); |
|||
assertThat(alarmService.findAlarmIdsByOriginatorId(tenantId, deviceId, 0, null, 10)).isNotEmpty(); |
|||
} |
|||
|
|||
private TsKvEntry getLatestTelemetry(EntityId entityId) throws Exception { |
|||
return timeseriesService.findLatest(tenantId, entityId, HousekeeperServiceTest.TELEMETRY_KEY).get().orElse(null); |
|||
} |
|||
|
|||
private List<TsKvEntry> getTimeseriesHistory(EntityId entityId) throws Exception { |
|||
return timeseriesService.findAll(tenantId, entityId, List.of(new BaseReadTsKvQuery(HousekeeperServiceTest.TELEMETRY_KEY, 0, System.currentTimeMillis(), 10, "DESC"))).get(); |
|||
} |
|||
|
|||
private List<EventInfo> getEvents(EntityId entityId) { |
|||
return eventService.findEvents(tenantId, entityId, EventType.LC_EVENT, new TimePageLink(100)).getData() |
|||
.stream().filter(event -> Optional.ofNullable(event.getBody()).map(body -> body.get("event")) |
|||
.map(JsonNode::asText).orElse("").equals("test")) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
|
|||
private Asset createAsset() { |
|||
Asset asset = new Asset(); |
|||
asset.setName("test"); |
|||
asset.setType("test"); |
|||
return doPost("/api/asset", asset, Asset.class); |
|||
} |
|||
|
|||
private RuleChainMetaData createRuleChain() { |
|||
RuleChain ruleChain = new RuleChain(); |
|||
ruleChain.setTenantId(tenantId); |
|||
ruleChain.setName("Test"); |
|||
ruleChain.setType(RuleChainType.CORE); |
|||
ruleChain.setDebugMode(true); |
|||
ruleChain.setConfiguration(JacksonUtil.newObjectNode().set("a", new TextNode("b"))); |
|||
ruleChain = ruleChainService.saveRuleChain(ruleChain); |
|||
RuleChainId ruleChainId = ruleChain.getId(); |
|||
|
|||
RuleChainMetaData metaData = new RuleChainMetaData(); |
|||
metaData.setRuleChainId(ruleChainId); |
|||
|
|||
RuleNode ruleNode1 = new RuleNode(); |
|||
ruleNode1.setName("Simple Rule Node 1"); |
|||
ruleNode1.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName()); |
|||
ruleNode1.setConfigurationVersion(TbGetAttributesNode.class.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class).version()); |
|||
ruleNode1.setDebugMode(true); |
|||
TbGetAttributesNodeConfiguration configuration1 = new TbGetAttributesNodeConfiguration(); |
|||
configuration1.setServerAttributeNames(Collections.singletonList("serverAttributeKey1")); |
|||
ruleNode1.setConfiguration(JacksonUtil.valueToTree(configuration1)); |
|||
|
|||
RuleNode ruleNode2 = new RuleNode(); |
|||
ruleNode2.setName("Simple Rule Node 2"); |
|||
ruleNode2.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName()); |
|||
ruleNode2.setConfigurationVersion(TbGetAttributesNode.class.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class).version()); |
|||
ruleNode2.setDebugMode(true); |
|||
TbGetAttributesNodeConfiguration configuration2 = new TbGetAttributesNodeConfiguration(); |
|||
configuration2.setServerAttributeNames(Collections.singletonList("serverAttributeKey2")); |
|||
ruleNode2.setConfiguration(JacksonUtil.valueToTree(configuration2)); |
|||
|
|||
metaData.setNodes(Arrays.asList(ruleNode1, ruleNode2)); |
|||
metaData.setFirstNodeIndex(0); |
|||
metaData.addConnectionInfo(0, 1, TbNodeConnectionType.SUCCESS); |
|||
ruleChainService.saveRuleChainMetaData(tenantId, metaData, Function.identity()); |
|||
return ruleChainService.loadRuleChainMetaData(tenantId, ruleChainId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import lombok.AccessLevel; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@NoArgsConstructor(access = AccessLevel.PROTECTED) |
|||
public class AlarmsDeletionHousekeeperTask extends HousekeeperTask { |
|||
|
|||
private List<UUID> alarms; |
|||
|
|||
public AlarmsDeletionHousekeeperTask(TenantId tenantId, EntityId entityId) { |
|||
this(tenantId, entityId, null); |
|||
} |
|||
|
|||
public AlarmsDeletionHousekeeperTask(TenantId tenantId, EntityId entityId, List<UUID> alarms) { |
|||
super(tenantId, entityId, HousekeeperTaskType.DELETE_ALARMS); |
|||
this.alarms = alarms; |
|||
} |
|||
|
|||
@Override |
|||
public String getDescription() { |
|||
return super.getDescription() + (alarms != null ? " (" + alarms + ")" : ""); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import lombok.AccessLevel; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.User; |
|||
|
|||
@Data |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@NoArgsConstructor(access = AccessLevel.PROTECTED) |
|||
public class AlarmsUnassignHousekeeperTask extends HousekeeperTask { |
|||
|
|||
private String userTitle; |
|||
|
|||
protected AlarmsUnassignHousekeeperTask(User user) { |
|||
super(user.getTenantId(), user.getId(), HousekeeperTaskType.UNASSIGN_ALARMS); |
|||
this.userTitle = user.getTitle(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@NoArgsConstructor |
|||
public class EntitiesDeletionHousekeeperTask extends HousekeeperTask { |
|||
|
|||
private EntityType entityType; |
|||
private List<UUID> entities; |
|||
|
|||
public EntitiesDeletionHousekeeperTask(TenantId tenantId, EntityType entityType, List<UUID> entities) { |
|||
super(tenantId, tenantId, HousekeeperTaskType.DELETE_ENTITIES); |
|||
this.entityType = entityType; |
|||
this.entities = entities; |
|||
} |
|||
|
|||
@JsonIgnore |
|||
@Override |
|||
public String getDescription() { |
|||
return entityType.getNormalName().toLowerCase() + "s deletion (" + entities + ")"; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,89 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
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 lombok.AccessLevel; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.NonNull; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "taskType", visible = true, include = JsonTypeInfo.As.EXISTING_PROPERTY, defaultImpl = HousekeeperTask.class) |
|||
@JsonSubTypes({ |
|||
@Type(name = "DELETE_TS_HISTORY", value = TsHistoryDeletionHousekeeperTask.class), |
|||
@Type(name = "DELETE_LATEST_TS", value = LatestTsDeletionHousekeeperTask.class), |
|||
@Type(name = "DELETE_TENANT_ENTITIES", value = TenantEntitiesDeletionHousekeeperTask.class), |
|||
@Type(name = "DELETE_ENTITIES", value = EntitiesDeletionHousekeeperTask.class), |
|||
@Type(name = "DELETE_ALARMS", value = AlarmsDeletionHousekeeperTask.class), |
|||
@Type(name = "UNASSIGN_ALARMS", value = AlarmsUnassignHousekeeperTask.class) |
|||
}) |
|||
@Data |
|||
@NoArgsConstructor(access = AccessLevel.PROTECTED) |
|||
public class HousekeeperTask implements Serializable { |
|||
|
|||
private TenantId tenantId; |
|||
private EntityId entityId; |
|||
private HousekeeperTaskType taskType; |
|||
private long ts; |
|||
|
|||
protected HousekeeperTask(@NonNull TenantId tenantId, @NonNull EntityId entityId, @NonNull HousekeeperTaskType taskType) { |
|||
this.tenantId = tenantId; |
|||
this.entityId = entityId; |
|||
this.taskType = taskType; |
|||
this.ts = System.currentTimeMillis(); |
|||
} |
|||
|
|||
public static HousekeeperTask deleteAttributes(TenantId tenantId, EntityId entityId) { |
|||
return new HousekeeperTask(tenantId, entityId, HousekeeperTaskType.DELETE_ATTRIBUTES); |
|||
} |
|||
|
|||
public static HousekeeperTask deleteTelemetry(TenantId tenantId, EntityId entityId) { |
|||
return new HousekeeperTask(tenantId, entityId, HousekeeperTaskType.DELETE_TELEMETRY); |
|||
} |
|||
|
|||
public static HousekeeperTask deleteEvents(TenantId tenantId, EntityId entityId) { |
|||
return new HousekeeperTask(tenantId, entityId, HousekeeperTaskType.DELETE_EVENTS); |
|||
} |
|||
|
|||
public static HousekeeperTask unassignAlarms(User user) { |
|||
return new AlarmsUnassignHousekeeperTask(user); |
|||
} |
|||
|
|||
public static HousekeeperTask deleteAlarms(TenantId tenantId, EntityId entityId) { |
|||
return new AlarmsDeletionHousekeeperTask(tenantId, entityId); |
|||
} |
|||
|
|||
public static HousekeeperTask deleteTenantEntities(TenantId tenantId, EntityType entityType) { |
|||
return new TenantEntitiesDeletionHousekeeperTask(tenantId, entityType); |
|||
} |
|||
|
|||
@JsonIgnore |
|||
public String getDescription() { |
|||
return taskType.getDescription() + " for " + entityId.getEntityType().getNormalName().toLowerCase() + " " + entityId.getId(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,37 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.RequiredArgsConstructor; |
|||
|
|||
@RequiredArgsConstructor |
|||
@Getter |
|||
public enum HousekeeperTaskType { |
|||
|
|||
DELETE_ATTRIBUTES("attributes deletion"), |
|||
DELETE_TELEMETRY("telemetry deletion"), |
|||
DELETE_LATEST_TS("latest telemetry deletion"), |
|||
DELETE_TS_HISTORY("timeseries history deletion"), |
|||
DELETE_EVENTS("events deletion"), |
|||
DELETE_ALARMS("alarms deletion"), |
|||
UNASSIGN_ALARMS("alarms unassigning"), |
|||
DELETE_TENANT_ENTITIES("tenant entities deletion"), |
|||
DELETE_ENTITIES("entities deletion"); |
|||
|
|||
private final String description; |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import lombok.AccessLevel; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
@Data |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@NoArgsConstructor(access = AccessLevel.PROTECTED) |
|||
public class LatestTsDeletionHousekeeperTask extends HousekeeperTask { |
|||
|
|||
private String key; |
|||
|
|||
public LatestTsDeletionHousekeeperTask(TenantId tenantId, EntityId entityId, String key) { |
|||
super(tenantId, entityId, HousekeeperTaskType.DELETE_LATEST_TS); |
|||
this.key = key; |
|||
} |
|||
|
|||
@Override |
|||
public String getDescription() { |
|||
return super.getDescription() + (key != null ? " for key '" + key + "'" : ""); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
@Data |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@NoArgsConstructor |
|||
public class TenantEntitiesDeletionHousekeeperTask extends HousekeeperTask { |
|||
|
|||
private EntityType entityType; |
|||
|
|||
public TenantEntitiesDeletionHousekeeperTask(TenantId tenantId, EntityType entityType) { |
|||
super(tenantId, tenantId, HousekeeperTaskType.DELETE_TENANT_ENTITIES); |
|||
this.entityType = entityType; |
|||
} |
|||
|
|||
@JsonIgnore |
|||
@Override |
|||
public String getDescription() { |
|||
return entityType.getNormalName().toLowerCase() + "s deletion"; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.housekeeper; |
|||
|
|||
import lombok.AccessLevel; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
@Data |
|||
@ToString(callSuper = true) |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@NoArgsConstructor(access = AccessLevel.PROTECTED) |
|||
public class TsHistoryDeletionHousekeeperTask extends HousekeeperTask { |
|||
|
|||
private String key; |
|||
|
|||
public TsHistoryDeletionHousekeeperTask(TenantId tenantId, EntityId entityId, String key) { |
|||
super(tenantId, entityId, HousekeeperTaskType.DELETE_TS_HISTORY); |
|||
this.key = key; |
|||
} |
|||
|
|||
@Override |
|||
public String getDescription() { |
|||
return super.getDescription() + (key != null ? " for key '" + key + "'" : ""); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,61 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.notification.info; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.Map; |
|||
|
|||
import static org.thingsboard.server.common.data.util.CollectionsUtil.mapOf; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class TaskProcessingFailureNotificationInfo implements RuleOriginatedNotificationInfo { |
|||
|
|||
private TenantId tenantId; |
|||
private EntityId entityId; |
|||
private HousekeeperTaskType taskType; |
|||
private String taskDescription; |
|||
private String error; |
|||
private int attempt; |
|||
|
|||
@Override |
|||
public Map<String, String> getTemplateData() { |
|||
return mapOf( |
|||
"tenantId", tenantId.toString(), |
|||
"entityType", entityId.getEntityType().getNormalName(), |
|||
"entityId", entityId.getId().toString(), |
|||
"taskType", taskType.getDescription(), |
|||
"taskDescription", taskDescription, |
|||
"error", error, |
|||
"attempt", String.valueOf(attempt) |
|||
); |
|||
} |
|||
|
|||
@Override |
|||
public TenantId getAffectedTenantId() { |
|||
return tenantId; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,53 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.notification.rule.trigger; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
|
|||
@Data |
|||
@Builder |
|||
public class TaskProcessingFailureTrigger implements NotificationRuleTrigger { |
|||
|
|||
private final HousekeeperTask task; |
|||
private final int attempt; |
|||
private final Throwable error; |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getType() { |
|||
return NotificationRuleTriggerType.TASK_PROCESSING_FAILURE; |
|||
} |
|||
|
|||
@Override |
|||
public TenantId getTenantId() { |
|||
return task.getTenantId(); |
|||
} |
|||
|
|||
@Override |
|||
public EntityId getOriginatorEntityId() { |
|||
return task.getEntityId(); |
|||
} |
|||
|
|||
@Override |
|||
public boolean deduplicate() { |
|||
return false; |
|||
} |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue