YevhenBondarenko
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